mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-08-03 16:39:00 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

bpo-32357: Optimize asyncio.iscoroutine() for non-native coroutines (#4915)

Browse source
This commit is contained in:
Yury Selivanov 2017-12-19 07:18:45 -05:00 • committed by GitHub
parent a7bd64c0c0
commit a9d7e552c7
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 149 additions and 33 deletions
Download patch file Download diff file Expand all files Collapse all files

43
Lib/test/test_asyncio/test_tasks.py
View file

@ -62,6 +62,20 @@ class Dummy:
pass
class CoroLikeObject:
def send(self, v):
raise StopIteration(42)
def throw(self, *exc):
pass
def close(self):
pass
def __await__(self):
return self
class BaseTaskTests:
Task = None
@ -2085,6 +2099,12 @@ class BaseTaskTests:
"a coroutine was expected, got 123"):
self.new_task(self.loop, 123)
# test it for the second time to ensure that caching
# in asyncio.iscoroutine() doesn't break things.
with self.assertRaisesRegex(TypeError,
"a coroutine was expected, got 123"):
self.new_task(self.loop, 123)
def test_create_task_with_oldstyle_coroutine(self):
@asyncio.coroutine
@ -2095,6 +2115,12 @@ class BaseTaskTests:
self.assertIsInstance(task, self.Task)
self.loop.run_until_complete(task)
# test it for the second time to ensure that caching
# in asyncio.iscoroutine() doesn't break things.
task = self.new_task(self.loop, coro())
self.assertIsInstance(task, self.Task)
self.loop.run_until_complete(task)
def test_create_task_with_async_function(self):
async def coro():
@ -2104,6 +2130,23 @@ class BaseTaskTests:
self.assertIsInstance(task, self.Task)
self.loop.run_until_complete(task)
# test it for the second time to ensure that caching
# in asyncio.iscoroutine() doesn't break things.
task = self.new_task(self.loop, coro())
self.assertIsInstance(task, self.Task)
self.loop.run_until_complete(task)
def test_create_task_with_asynclike_function(self):
task = self.new_task(self.loop, CoroLikeObject())
self.assertIsInstance(task, self.Task)
self.assertEqual(self.loop.run_until_complete(task), 42)
# test it for the second time to ensure that caching
# in asyncio.iscoroutine() doesn't break things.
task = self.new_task(self.loop, CoroLikeObject())
self.assertIsInstance(task, self.Task)
self.assertEqual(self.loop.run_until_complete(task), 42)
def test_bare_create_task(self):
async def inner():