+import asyncio
import builtins
import locale
import logging
'sysconfig._CONFIG_VARS', 'sysconfig._INSTALL_SCHEMES',
'files', 'locale', 'warnings.showwarning',
'shutil_archive_formats', 'shutil_unpack_formats',
+ 'asyncio.events._event_loop_policy',
)
+ def get_asyncio_events__event_loop_policy(self):
+ return support.maybe_get_event_loop_policy()
+ def restore_asyncio_events__event_loop_policy(self, policy):
+ asyncio.set_event_loop_policy(policy)
+
def get_sys_argv(self):
return id(sys.argv), sys.argv, sys.argv[:]
def restore_sys_argv(self, saved_argv):
if __name__ != 'test.support':
raise ImportError('support must be imported from the test package')
+import asyncio.events
import collections.abc
import contextlib
import errno
raise self.path
else:
return self.path
+
+
+def maybe_get_event_loop_policy():
+ """Return the global event loop policy if one is set, else return None."""
+ return asyncio.events._event_loop_policy
def tearDown(self):
self.loop.close()
self.loop = None
+ asyncio.set_event_loop_policy(None)
async def to_list(self, gen):
res = []
PY34 = sys.version_info >= (3, 4)
+def tearDownModule():
+ asyncio.set_event_loop_policy(None)
+
+
def mock_socket_module():
m_socket = mock.MagicMock(spec=socket)
for name in (
from test.test_asyncio import functional as func_tests
+def tearDownModule():
+ asyncio.set_event_loop_policy(None)
+
+
class ReceiveStuffProto(asyncio.BufferedProtocol):
def __init__(self, cb, con_lost_fut):
self.cb = cb
import unittest
+def tearDownModule():
+ asyncio.set_event_loop_policy(None)
+
+
class DecimalContextTest(unittest.TestCase):
def test_asyncio_task_decimal_context(self):
from test import support
+def tearDownModule():
+ asyncio.set_event_loop_policy(None)
+
+
def osx_tiger():
"""Return True if the platform is Mac OS 10.4 or older."""
if sys.platform != 'darwin':
from test import support
+def tearDownModule():
+ asyncio.set_event_loop_policy(None)
+
+
def _fakefunc(f):
return f
RGX_REPR = re.compile(STR_RGX_REPR)
+def tearDownModule():
+ asyncio.set_event_loop_policy(None)
+
+
class LockTests(test_utils.TestCase):
def setUp(self):
from test.test_asyncio import utils as test_utils
+def tearDownModule():
+ asyncio.set_event_loop_policy(None)
+
+
# Test that asyncio.iscoroutine() uses collections.abc.Coroutine
class FakeCoro:
def send(self, value):
from test.test_asyncio import utils as test_utils
+def tearDownModule():
+ asyncio.set_event_loop_policy(None)
+
+
def close_transport(transport):
# Don't call transport.close() because the event loop and the IOCP proactor
# are mocked
from test.test_asyncio import utils as test_utils
+def tearDownModule():
+ asyncio.set_event_loop_policy(None)
+
+
class _QueueTestBase(test_utils.TestCase):
def setUp(self):
MOCK_ANY = mock.ANY
+def tearDownModule():
+ asyncio.set_event_loop_policy(None)
+
+
class TestBaseSelectorEventLoop(BaseSelectorEventLoop):
def _make_self_pipe(self):
from test.test_asyncio import functional as func_tests
+def tearDownModule():
+ asyncio.set_event_loop_policy(None)
+
+
class BaseStartServer(func_tests.FunctionalTestCaseMixin):
def new_loop(self):
from test.test_asyncio import functional as func_tests
+def tearDownModule():
+ asyncio.set_event_loop_policy(None)
+
+
@unittest.skipIf(ssl is None, 'No ssl module')
class SslProtoHandshakeTests(test_utils.TestCase):
from test.test_asyncio import utils as test_utils
+def tearDownModule():
+ asyncio.set_event_loop_policy(None)
+
+
class StreamTests(test_utils.TestCase):
DATA = b'line1\nline2\nline3\n'
'data = sys.stdin.buffer.read()',
'sys.stdout.buffer.write(data)'))]
+
+def tearDownModule():
+ asyncio.set_event_loop_policy(None)
+
+
class TestSubprocessTransport(base_subprocess.BaseSubprocessTransport):
def _start(self, *args, **kwargs):
self._proc = mock.Mock()
from test.support.script_helper import assert_python_ok
+def tearDownModule():
+ asyncio.set_event_loop_policy(None)
+
+
@asyncio.coroutine
def coroutine_function():
pass
MOCK_ANY = mock.ANY
+def tearDownModule():
+ asyncio.set_event_loop_policy(None)
+
+
def close_pipe_transport(transport):
# Don't call transport.close() because the event loop and the selector
# are mocked
from test.test_asyncio import utils as test_utils
+def tearDownModule():
+ asyncio.set_event_loop_policy(None)
+
+
class UpperProto(asyncio.Protocol):
def __init__(self):
self.buf = []
import _overlapped
import _winapi
+import asyncio
from asyncio import windows_utils
from test import support
+def tearDownModule():
+ asyncio.set_event_loop_policy(None)
+
+
class PipeTests(unittest.TestCase):
def test_pipe_overlapped(self):
return loop.run_until_complete(coro)
finally:
loop.close()
- asyncio.set_event_loop(None)
+ asyncio.set_event_loop_policy(None)
return wrapper
self.loop = asyncio.new_event_loop()
asyncio.set_event_loop(self.loop)
self.addCleanup(self.loop.close)
+ self.addCleanup(asyncio.set_event_loop_policy, None)
@_async_test
async def test_async_callback(self):
pass
finally:
loop.close()
- asyncio.set_event_loop(None)
+ asyncio.set_event_loop_policy(None)
self.assertEqual(buffer, [1, 2, 'MyException'])
... loop = asyncio.new_event_loop()
... loop.run_until_complete(test_main())
... loop.close()
+ ... 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)
... print("finished")
>>> with PdbTestInput(['step',
... loop = asyncio.new_event_loop()
... loop.run_until_complete(test_main())
... loop.close()
+ ... 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)
... print("finished")
>>> with PdbTestInput(['step',
with self.assertRaisesRegex(*error):
asyncio.run(func(output))
sys.settrace(None)
+ asyncio.set_event_loop_policy(None)
self.compare_jump_output(expected, output)
def jump_test(jumpFrom, jumpTo, expected, error=None, event='line'):
--- /dev/null
+Check that a global asyncio event loop policy is not left behind by any
+tests.
projects / python / commitdiff