mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-08-31 05:58:33 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

bpo-34790: Implement deprecation of passing coroutines to asyncio.wait() (GH-16977)

Browse source
This commit is contained in:
Kyle Stanley 2019-12-30 06:50:19 -05:00 • committed by Andrew Svetlov
parent 88dce26da6
commit 89aa7f0ede
3 changed files with 28 additions and 5 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -979,12 +979,12 @@ class BaseTaskTests:
def coro(s):
return s
c = coro('test')
task =self.new_task(
task = self.new_task(
self.loop,
asyncio.wait([c, c, coro('spam')]))
done, pending = self.loop.run_until_complete(task)
with self.assertWarns(DeprecationWarning):
done, pending = self.loop.run_until_complete(task)
self.assertFalse(pending)
self.assertEqual(set(f.result() for f in done), {'test', 'spam'})
@ -1346,7 +1346,9 @@ class BaseTaskTests:
futs = list(asyncio.as_completed(fs, loop=loop))
self.assertEqual(len(futs), 2)
waiter = asyncio.wait(futs)
done, pending = loop.run_until_complete(waiter)
# Deprecation from passing coros in futs to asyncio.wait()
with self.assertWarns(DeprecationWarning):
done, pending = loop.run_until_complete(waiter)
self.assertEqual(set(f.result() for f in done), {'a', 'b'})
def test_as_completed_duplicate_coroutines(self):
@ -1751,7 +1753,8 @@ class BaseTaskTests:
async def outer():
nonlocal proof
d, p = await asyncio.wait([inner()])
with self.assertWarns(DeprecationWarning):
d, p = await asyncio.wait([inner()])
proof += 100
f = asyncio.ensure_future(outer(), loop=self.loop)
@ -3307,6 +3310,17 @@ class WaitTests(test_utils.TestCase):
self.loop.run_until_complete(
asyncio.wait_for(coroutine_function(), 0.01, loop=self.loop))
def test_coro_is_deprecated_in_wait(self):
# Remove test when passing coros to asyncio.wait() is removed in 3.11
with self.assertWarns(DeprecationWarning):
self.loop.run_until_complete(
asyncio.wait([coroutine_function()]))
task = self.loop.create_task(coroutine_function())
with self.assertWarns(DeprecationWarning):
self.loop.run_until_complete(
asyncio.wait([task, coroutine_function()]))
class CompatibilityTests(test_utils.TestCase):
# Tests for checking a bridge between old-styled coroutines