mirror of
https://github.com/python/cpython.git
synced 2025-08-03 16:39:00 +00:00
Sync asyncio changes from the main repo.
This commit is contained in:
parent
37c4f78390
commit
740169cd24
5 changed files with 94 additions and 16 deletions
|
@ -16,10 +16,15 @@ from asyncio import constants
|
|||
from asyncio import test_utils
|
||||
try:
|
||||
from test import support
|
||||
from test.script_helper import assert_python_ok
|
||||
except ImportError:
|
||||
from asyncio import test_support as support
|
||||
from asyncio.test_support import assert_python_ok
|
||||
try:
|
||||
from test.support.script_helper import assert_python_ok
|
||||
except ImportError:
|
||||
try:
|
||||
from test.script_helper import assert_python_ok
|
||||
except ImportError:
|
||||
from asyncio.test_support import assert_python_ok
|
||||
|
||||
|
||||
MOCK_ANY = mock.ANY
|
||||
|
@ -623,6 +628,42 @@ class BaseEventLoopTests(test_utils.TestCase):
|
|||
self.assertIs(type(_context['context']['exception']),
|
||||
ZeroDivisionError)
|
||||
|
||||
def test_set_task_factory_invalid(self):
|
||||
with self.assertRaisesRegex(
|
||||
TypeError, 'task factory must be a callable or None'):
|
||||
|
||||
self.loop.set_task_factory(1)
|
||||
|
||||
self.assertIsNone(self.loop.get_task_factory())
|
||||
|
||||
def test_set_task_factory(self):
|
||||
self.loop._process_events = mock.Mock()
|
||||
|
||||
class MyTask(asyncio.Task):
|
||||
pass
|
||||
|
||||
@asyncio.coroutine
|
||||
def coro():
|
||||
pass
|
||||
|
||||
factory = lambda loop, coro: MyTask(coro, loop=loop)
|
||||
|
||||
self.assertIsNone(self.loop.get_task_factory())
|
||||
self.loop.set_task_factory(factory)
|
||||
self.assertIs(self.loop.get_task_factory(), factory)
|
||||
|
||||
task = self.loop.create_task(coro())
|
||||
self.assertTrue(isinstance(task, MyTask))
|
||||
self.loop.run_until_complete(task)
|
||||
|
||||
self.loop.set_task_factory(None)
|
||||
self.assertIsNone(self.loop.get_task_factory())
|
||||
|
||||
task = self.loop.create_task(coro())
|
||||
self.assertTrue(isinstance(task, asyncio.Task))
|
||||
self.assertFalse(isinstance(task, MyTask))
|
||||
self.loop.run_until_complete(task)
|
||||
|
||||
def test_env_var_debug(self):
|
||||
code = '\n'.join((
|
||||
'import asyncio',
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue