mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-08-31 14:07:50 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 9

bpo-31185: Fixed miscellaneous errors in asyncio speedup module. (#3076)

Browse source
This commit is contained in:
Serhiy Storchaka 2017-09-03 08:10:14 +03:00 • committed by GitHub
parent 8df44ee8e0
commit bca4939d80
4 changed files with 250 additions and 221 deletions
Download patch file Download diff file Expand all files Collapse all files

45
Lib/test/test_asyncio/test_futures.py
View file

@ -100,8 +100,8 @@ class DuckTests(test_utils.TestCase):
class BaseFutureTests:
def _new_future(self, loop=None):
raise NotImplementedError
def _new_future(self, *args, **kwargs):
return self.cls(*args, **kwargs)
def setUp(self):
super().setUp()
@ -147,6 +147,39 @@ class BaseFutureTests:
# Make sure Future doesn't accept a positional argument
self.assertRaises(TypeError, self._new_future, 42)
def test_uninitialized(self):
fut = self.cls.__new__(self.cls, loop=self.loop)
self.assertRaises(asyncio.InvalidStateError, fut.result)
fut = self.cls.__new__(self.cls, loop=self.loop)
self.assertRaises(asyncio.InvalidStateError, fut.exception)
fut = self.cls.__new__(self.cls, loop=self.loop)
with self.assertRaises((RuntimeError, AttributeError)):
fut.set_result(None)
fut = self.cls.__new__(self.cls, loop=self.loop)
with self.assertRaises((RuntimeError, AttributeError)):
fut.set_exception(Exception)
fut = self.cls.__new__(self.cls, loop=self.loop)
with self.assertRaises((RuntimeError, AttributeError)):
fut.cancel()
fut = self.cls.__new__(self.cls, loop=self.loop)
with self.assertRaises((RuntimeError, AttributeError)):
fut.add_done_callback(lambda f: None)
fut = self.cls.__new__(self.cls, loop=self.loop)
with self.assertRaises((RuntimeError, AttributeError)):
fut.remove_done_callback(lambda f: None)
fut = self.cls.__new__(self.cls, loop=self.loop)
with self.assertRaises((RuntimeError, AttributeError)):
fut._schedule_callbacks()
fut = self.cls.__new__(self.cls, loop=self.loop)
try:
repr(fut)
except AttributeError:
pass
fut = self.cls.__new__(self.cls, loop=self.loop)
fut.cancelled()
fut.done()
iter(fut)
def test_cancel(self):
f = self._new_future(loop=self.loop)
self.assertTrue(f.cancel())
@ -501,15 +534,11 @@ class BaseFutureTests:
@unittest.skipUnless(hasattr(futures, '_CFuture'),
'requires the C _asyncio module')
class CFutureTests(BaseFutureTests, test_utils.TestCase):
def _new_future(self, *args, **kwargs):
return futures._CFuture(*args, **kwargs)
cls = getattr(futures, '_CFuture')
class PyFutureTests(BaseFutureTests, test_utils.TestCase):
def _new_future(self, *args, **kwargs):
return futures._PyFuture(*args, **kwargs)
cls = futures._PyFuture
class BaseFutureDoneCallbackTests():