traverseda
/
cpython
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

(Merge 3.4) Issue #21163: Fix "destroy pending task" warning in test_wait_errors()

This commit is contained in:
Victor Stinner 2014-07-16 18:50:54 +02:00
parent 845212af41 e931f7b768
commit f85581f6e5
2 changed files with 9 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
Lib/asyncio/tasks.py
View File

@ -330,14 +330,14 @@ def wait(fs, *, loop=None, timeout=None, return_when=ALL_COMPLETED):
raise TypeError("expect a list of futures, not %s" % type(fs).__name__) raise TypeError("expect a list of futures, not %s" % type(fs).__name__)
if not fs: if not fs:
raise ValueError('Set of coroutines/Futures is empty.') raise ValueError('Set of coroutines/Futures is empty.')
if return_when not in (FIRST_COMPLETED, FIRST_EXCEPTION, ALL_COMPLETED):
raise ValueError('Invalid return_when value: {}'.format(return_when))
if loop is None: if loop is None:
loop = events.get_event_loop() loop = events.get_event_loop()
fs = {async(f, loop=loop) for f in set(fs)} fs = {async(f, loop=loop) for f in set(fs)}
if return_when not in (FIRST_COMPLETED, FIRST_EXCEPTION, ALL_COMPLETED):
raise ValueError('Invalid return_when value: {}'.format(return_when))
return (yield from _wait(fs, timeout, return_when, loop)) return (yield from _wait(fs, timeout, return_when, loop))

11
Lib/test/test_asyncio/test_tasks.py
View File

@ -623,10 +623,13 @@ class TaskTests(test_utils.TestCase):
ValueError, self.loop.run_until_complete, ValueError, self.loop.run_until_complete,
asyncio.wait(set(), loop=self.loop)) asyncio.wait(set(), loop=self.loop))
self.assertRaises( # -1 is an invalid return_when value
ValueError, self.loop.run_until_complete, sleep_coro = asyncio.sleep(10.0, loop=self.loop)
asyncio.wait([asyncio.sleep(10.0, loop=self.loop)], wait_coro = asyncio.wait([sleep_coro], return_when=-1, loop=self.loop)
return_when=-1, loop=self.loop)) self.assertRaises(ValueError,
self.loop.run_until_complete, wait_coro)
sleep_coro.close()
def test_wait_first_completed(self): def test_wait_first_completed(self):