"""Tests for futures.py.""" import concurrent.futures import gc import re import sys import threading import unittest from unittest import mock import asyncio from asyncio import futures from test.test_asyncio import utils as test_utils from test import support def _fakefunc(f): return f def first_cb(): pass def last_cb(): pass class DuckFuture: # Class that does not inherit from Future but aims to be duck-type # compatible with it. _asyncio_future_blocking = False __cancelled = False __result = None __exception = None def cancel(self): if self.done(): return False self.__cancelled = True return True def cancelled(self): return self.__cancelled def done(self): return (self.__cancelled or self.__result is not None or self.__exception is not None) def result(self): assert not self.cancelled() if self.__exception is not None: raise self.__exception return self.__result def exception(self): assert not self.cancelled() return self.__exception def set_result(self, result): assert not self.done() assert result is not None self.__result = result def set_exception(self, exception): assert not self.done() assert exception is not None self.__exception = exception def __iter__(self): if not self.done(): self._asyncio_future_blocking = True yield self assert self.done() return self.result() class DuckTests(test_utils.TestCase): def setUp(self): super().setUp() self.loop = self.new_test_loop() self.addCleanup(self.loop.close) def test_wrap_future(self): f = DuckFuture() g = asyncio.wrap_future(f) assert g is f def test_ensure_future(self): f = DuckFuture() g = asyncio.ensure_future(f) assert g is f class BaseFutureTests: def _new_future(self, *args, **kwargs): return self.cls(*args, **kwargs) def setUp(self): super().setUp() self.loop = self.new_test_loop() self.addCleanup(self.loop.close) def test_isfuture(self): class MyFuture: _asyncio_future_blocking = None def __init__(self): self._asyncio_future_blocking = False self.assertFalse(asyncio.isfuture(MyFuture)) self.assertTrue(asyncio.isfuture(MyFuture())) self.assertFalse(asyncio.isfuture(1)) # As `isinstance(Mock(), Future)` returns `False` self.assertFalse(asyncio.isfuture(mock.Mock())) f = self._new_future(loop=self.loop) self.assertTrue(asyncio.isfuture(f)) self.assertFalse(asyncio.isfuture(type(f))) # As `isinstance(Mock(Future), Future)` returns `True` self.assertTrue(asyncio.isfuture(mock.Mock(type(f)))) f.cancel() def test_initial_state(self): f = self._new_future(loop=self.loop) self.assertFalse(f.cancelled()) self.assertFalse(f.done()) f.cancel() self.assertTrue(f.cancelled()) def test_init_constructor_default_loop(self): asyncio.set_event_loop(self.loop) f = self._new_future() self.assertIs(f._loop, self.loop) self.assertIs(f.get_loop(), self.loop) def test_constructor_positional(self): # Make sure Future doesn't accept a positional argument self.assertRaises(TypeError, self._new_future, 42) def test_uninitialized(self): # Test that C Future doesn't crash when Future.__init__() # call was skipped. 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) try: repr(fut) except (RuntimeError, AttributeError): pass fut = self.cls.__new__(self.cls, loop=self.loop) try: fut.__await__() except RuntimeError: pass fut = self.cls.__new__(self.cls, loop=self.loop) try: iter(fut) except RuntimeError: pass fut = self.cls.__new__(self.cls, loop=self.loop) self.assertFalse(fut.cancelled()) self.assertFalse(fut.done()) def test_cancel(self): f = self._new_future(loop=self.loop) self.assertTrue(f.cancel()) self.assertTrue(f.cancelled()) self.assertTrue(f.done()) self.assertRaises(asyncio.CancelledError, f.result) self.assertRaises(asyncio.CancelledError, f.exception) self.assertRaises(asyncio.InvalidStateError, f.set_result, None) self.assertRaises(asyncio.InvalidStateError, f.set_exception, None) self.assertFalse(f.cancel()) def test_result(self): f = self._new_future(loop=self.loop) self.assertRaises(asyncio.InvalidStateError, f.result) f.set_result(42) self.assertFalse(f.cancelled()) self.assertTrue(f.done()) self.assertEqual(f.result(), 42) self.assertEqual(f.exception(), None) self.assertRaises(asyncio.InvalidStateError, f.set_result, None) self.assertRaises(asyncio.InvalidStateError, f.set_exception, None) self.assertFalse(f.cancel()) def test_exception(self): exc = RuntimeError() f = self._new_future(loop=self.loop) self.assertRaises(asyncio.InvalidStateError, f.exception) # StopIteration cannot be raised into a Future - CPython issue26221 self.assertRaisesRegex(TypeError, "StopIteration .* cannot be raised", f.set_exception, StopIteration) f.set_exception(exc) self.assertFalse(f.cancelled()) self.assertTrue(f.done()) self.assertRaises(RuntimeError, f.result) self.assertEqual(f.exception(), exc) self.assertRaises(asyncio.InvalidStateError, f.set_result, None) self.assertRaises(asyncio.InvalidStateError, f.set_exception, None) self.assertFalse(f.cancel()) def test_exception_class(self): f = self._new_future(loop=self.loop) f.set_exception(RuntimeError) self.assertIsInstance(f.exception(), RuntimeError) def test_yield_from_twice(self): f = self._new_future(loop=self.loop) def fixture(): yield 'A' x = yield from f yield 'B', x y = yield from f yield 'C', y g = fixture() self.assertEqual(next(g), 'A') # yield 'A'. self.assertEqual(next(g), f) # First yield from f. f.set_result(42) self.assertEqual(next(g), ('B', 42)) # yield 'B', x. # The second "yield from f" does not yield f. self.assertEqual(next(g), ('C', 42)) # yield 'C', y. def test_future_repr(self): self.loop.set_debug(True) f_pending_debug = self._new_future(loop=self.loop) frame = f_pending_debug._source_traceback[-1] self.assertEqual( repr(f_pending_debug), f'<{self.cls.__name__} pending created at {frame[0]}:{frame[1]}>') f_pending_debug.cancel() self.loop.set_debug(False) f_pending = self._new_future(loop=self.loop) self.assertEqual(repr(f_pending), f'<{self.cls.__name__} pending>') f_pending.cancel() f_cancelled = self._new_future(loop=self.loop) f_cancelled.cancel() self.assertEqual(repr(f_cancelled), f'<{self.cls.__name__} cancelled>') f_result = self._new_future(loop=self.loop) f_result.set_result(4) self.assertEqual( repr(f_result), f'<{self.cls.__name__} finished result=4>') self.assertEqual(f_result.result(), 4) exc = RuntimeError() f_exception = self._new_future(loop=self.loop) f_exception.set_exception(exc) self.assertEqual( repr(f_exception), f'<{self.cls.__name__} finished exception=RuntimeError()>') self.assertIs(f_exception.exception(), exc) def func_repr(func): filename, lineno = test_utils.get_function_source(func) text = '%s() at %s:%s' % (func.__qualname__, filename, lineno) return re.escape(text) f_one_callbacks = self._new_future(loop=self.loop) f_one_callbacks.add_done_callback(_fakefunc) fake_repr = func_repr(_fakefunc) self.assertRegex( repr(f_one_callbacks), r'<' + self.cls.__name__ + r' pending cb=\[%s\]>' % fake_repr) f_one_callbacks.cancel() self.assertEqual(repr(f_one_callbacks), f'<{self.cls.__name__} cancelled>') f_two_callbacks = self._new_future(loop=self.loop) f_two_callbacks.add_done_callback(first_cb) f_two_callbacks.add_done_callback(last_cb) first_repr = func_repr(first_cb) last_repr = func_repr(last_cb) self.assertRegex(repr(f_two_callbacks), r'<' + self.cls.__name__ + r' pending cb=\[%s, %s\]>' % (first_repr, last_repr)) f_many_callbacks = self._new_future(loop=self.loop) f_many_callbacks.add_done_callback(first_cb) for i in range(8): f_many_callbacks.add_done_callback(_fakefunc) f_many_callbacks.add_done_callback(last_cb) cb_regex = r'%s, <8 more>, %s' % (first_repr, last_repr) self.assertRegex( repr(f_many_callbacks), r'<' + self.cls.__name__ + r' pending cb=\[%s\]>' % cb_regex) f_many_callbacks.cancel() self.assertEqual(repr(f_many_callbacks), f'<{self.cls.__name__} cancelled>') def test_copy_state(self): from asyncio.futures import _copy_future_state f = self._new_future(loop=self.loop) f.set_result(10) newf = self._new_future(loop=self.loop) _copy_future_state(f, newf) self.assertTrue(newf.done()) self.assertEqual(newf.result(), 10) f_exception = self._new_future(loop=self.loop) f_exception.set_exception(RuntimeError()) newf_exception = self._new_future(loop=self.loop) _copy_future_state(f_exception, newf_exception) self.assertTrue(newf_exception.done()) self.assertRaises(RuntimeError, newf_exception.result) f_cancelled = self._new_future(loop=self.loop) f_cancelled.cancel() newf_cancelled = self._new_future(loop=self.loop) _copy_future_state(f_cancelled, newf_cancelled) self.assertTrue(newf_cancelled.cancelled()) def test_iter(self): fut = self._new_future(loop=self.loop) def coro(): yield from fut def test(): arg1, arg2 = coro() with self.assertRaisesRegex(RuntimeError, "await wasn't used"): test() fut.cancel() def test_log_traceback(self): fut = self._new_future(loop=self.loop) with self.assertRaisesRegex(ValueError, 'can only be set to False'): fut._log_traceback = True @mock.patch('asyncio.base_events.logger') def test_tb_logger_abandoned(self, m_log): fut = self._new_future(loop=self.loop) del fut self.assertFalse(m_log.error.called) @mock.patch('asyncio.base_events.logger') def test_tb_logger_not_called_after_cancel(self, m_log): fut = self._new_future(loop=self.loop) fut.set_exception(Exception()) fut.cancel() del fut self.assertFalse(m_log.error.called) @mock.patch('asyncio.base_events.logger') def test_tb_logger_result_unretrieved(self, m_log): fut = self._new_future(loop=self.loop) fut.set_result(42) del fut self.assertFalse(m_log.error.called) @mock.patch('asyncio.base_events.logger') def test_tb_logger_result_retrieved(self, m_log): fut = self._new_future(loop=self.loop) fut.set_result(42) fut.result() del fut self.assertFalse(m_log.error.called) @mock.patch('asyncio.base_events.logger') def test_tb_logger_exception_unretrieved(self, m_log): fut = self._new_future(loop=self.loop) fut.set_exception(RuntimeError('boom')) del fut test_utils.run_briefly(self.loop) support.gc_collect() self.assertTrue(m_log.error.called) @mock.patch('asyncio.base_events.logger') def test_tb_logger_exception_retrieved(self, m_log): fut = self._new_future(loop=self.loop) fut.set_exception(RuntimeError('boom')) fut.exception() del fut self.assertFalse(m_log.error.called) @mock.patch('asyncio.base_events.logger') def test_tb_logger_exception_result_retrieved(self, m_log): fut = self._new_future(loop=self.loop) fut.set_exception(RuntimeError('boom')) self.assertRaises(RuntimeError, fut.result) del fut self.assertFalse(m_log.error.called) def test_wrap_future(self): def run(arg): return (arg, threading.get_ident()) ex = concurrent.futures.ThreadPoolExecutor(1) f1 = ex.submit(run, 'oi') f2 = asyncio.wrap_future(f1, loop=self.loop) res, ident = self.loop.run_until_complete(f2) self.assertTrue(asyncio.isfuture(f2)) self.assertEqual(res, 'oi') self.assertNotEqual(ident, threading.get_ident()) ex.shutdown(wait=True) def test_wrap_future_future(self): f1 = self._new_future(loop=self.loop) f2 = asyncio.wrap_future(f1) self.assertIs(f1, f2) def test_wrap_future_use_global_loop(self): with mock.patch('asyncio.futures.events') as events: events.get_event_loop = lambda: self.loop def run(arg): return (arg, threading.get_ident()) ex = concurrent.futures.ThreadPoolExecutor(1) f1 = ex.submit(run, 'oi') f2 = asyncio.wrap_future(f1) self.assertIs(self.loop, f2._loop) ex.shutdown(wait=True) def test_wrap_future_cancel(self): f1 = concurrent.futures.Future() f2 = asyncio.wrap_future(f1, loop=self.loop) f2.cancel() test_utils.run_briefly(self.loop) self.assertTrue(f1.cancelled()) self.assertTrue(f2.cancelled()) def test_wrap_future_cancel2(self): f1 = concurrent.futures.Future() f2 = asyncio.wrap_future(f1, loop=self.loop) f1.set_result(42) f2.cancel() test_utils.run_briefly(self.loop) self.assertFalse(f1.cancelled()) self.assertEqual(f1.result(), 42) self.assertTrue(f2.cancelled()) def test_future_source_traceback(self): self.loop.set_debug(True) future = self._new_future(loop=self.loop) lineno = sys._getframe().f_lineno - 1 self.assertIsInstance(future._source_traceback, list) self.assertEqual(future._source_traceback[-2][:3], (__file__, lineno, 'test_future_source_traceback')) @mock.patch('asyncio.base_events.logger') def check_future_exception_never_retrieved(self, debug, m_log): self.loop.set_debug(debug) def memory_error(): try: raise MemoryError() except BaseException as exc: return exc exc = memory_error() future = self._new_future(loop=self.loop) future.set_exception(exc) future = None test_utils.run_briefly(self.loop) support.gc_collect() if sys.version_info >= (3, 4): regex = f'^{self.cls.__name__} exception was never retrieved\n' exc_info = (type(exc), exc, exc.__traceback__) m_log.error.assert_called_once_with(mock.ANY, exc_info=exc_info) else: regex = r'^Future/Task exception was never retrieved\n' m_log.error.assert_called_once_with(mock.ANY, exc_info=False) message = m_log.error.call_args[0][0] self.assertRegex(message, re.compile(regex, re.DOTALL)) def test_future_exception_never_retrieved(self): self.check_future_exception_never_retrieved(False) def test_future_exception_never_retrieved_debug(self): self.check_future_exception_never_retrieved(True) def test_set_result_unless_cancelled(self): fut = self._new_future(loop=self.loop) fut.cancel() futures._set_result_unless_cancelled(fut, 2) self.assertTrue(fut.cancelled()) def test_future_stop_iteration_args(self): fut = self._new_future(loop=self.loop) fut.set_result((1, 2)) fi = fut.__iter__() result = None try: fi.send(None) except StopIteration as ex: result = ex.args[0] else: self.fail('StopIteration was expected') self.assertEqual(result, (1, 2)) def test_future_iter_throw(self): fut = self._new_future(loop=self.loop) fi = iter(fut) self.assertRaises(TypeError, fi.throw, Exception, Exception("elephant"), 32) self.assertRaises(TypeError, fi.throw, Exception("elephant"), Exception("elephant")) self.assertRaises(TypeError, fi.throw, list) def test_future_del_collect(self): class Evil: def __del__(self): gc.collect() for i in range(100): fut = self._new_future(loop=self.loop) fut.set_result(Evil()) @unittest.skipUnless(hasattr(futures, '_CFuture'), 'requires the C _asyncio module') class CFutureTests(BaseFutureTests, test_utils.TestCase): try: cls = futures._CFuture except AttributeError: cls = None @unittest.skipUnless(hasattr(futures, '_CFuture'), 'requires the C _asyncio module') class CSubFutureTests(BaseFutureTests, test_utils.TestCase): try: class CSubFuture(futures._CFuture): pass cls = CSubFuture except AttributeError: cls = None class PyFutureTests(BaseFutureTests, test_utils.TestCase): cls = futures._PyFuture class BaseFutureDoneCallbackTests(): def setUp(self): super().setUp() self.loop = self.new_test_loop() def run_briefly(self): test_utils.run_briefly(self.loop) def _make_callback(self, bag, thing): # Create a callback function that appends thing to bag. def bag_appender(future): bag.append(thing) return bag_appender def _new_future(self): raise NotImplementedError def test_callbacks_remove_first_callback(self): bag = [] f = self._new_future() cb1 = self._make_callback(bag, 42) cb2 = self._make_callback(bag, 17) cb3 = self._make_callback(bag, 100) f.add_done_callback(cb1) f.add_done_callback(cb2) f.add_done_callback(cb3) f.remove_done_callback(cb1) f.remove_done_callback(cb1) self.assertEqual(bag, []) f.set_result('foo') self.run_briefly() self.assertEqual(bag, [17, 100]) self.assertEqual(f.result(), 'foo') def test_callbacks_remove_first_and_second_callback(self): bag = [] f = self._new_future() cb1 = self._make_callback(bag, 42) cb2 = self._make_callback(bag, 17) cb3 = self._make_callback(bag, 100) f.add_done_callback(cb1) f.add_done_callback(cb2) f.add_done_callback(cb3) f.remove_done_callback(cb1) f.remove_done_callback(cb2) f.remove_done_callback(cb1) self.assertEqual(bag, []) f.set_result('foo') self.run_briefly() self.assertEqual(bag, [100]) self.assertEqual(f.result(), 'foo') def test_callbacks_remove_third_callback(self): bag = [] f = self._new_future() cb1 = self._make_callback(bag, 42) cb2 = self._make_callback(bag, 17) cb3 = self._make_callback(bag, 100) f.add_done_callback(cb1) f.add_done_callback(cb2) f.add_done_callback(cb3) f.remove_done_callback(cb3) f.remove_done_callback(cb3) self.assertEqual(bag, []) f.set_result('foo') self.run_briefly() self.assertEqual(bag, [42, 17]) self.assertEqual(f.result(), 'foo') def test_callbacks_invoked_on_set_result(self): bag = [] f = self._new_future() f.add_done_callback(self._make_callback(bag, 42)) f.add_done_callback(self._make_callback(bag, 17)) self.assertEqual(bag, []) f.set_result('foo') self.run_briefly() self.assertEqual(bag, [42, 17]) self.assertEqual(f.result(), 'foo') def test_callbacks_invoked_on_set_exception(self): bag = [] f = self._new_future() f.add_done_callback(self._make_callback(bag, 100)) self.assertEqual(bag, []) exc = RuntimeError() f.set_exception(exc) self.run_briefly() self.assertEqual(bag, [100]) self.assertEqual(f.exception(), exc) def test_remove_done_callback(self): bag = [] f = self._new_future() cb1 = self._make_callback(bag, 1) cb2 = self._make_callback(bag, 2) cb3 = self._make_callback(bag, 3) # Add one cb1 and one cb2. f.add_done_callback(cb1) f.add_done_callback(cb2) # One instance of cb2 removed. Now there's only one cb1. self.assertEqual(f.remove_done_callback(cb2), 1) # Never had any cb3 in there. self.assertEqual(f.remove_done_callback(cb3), 0) # After this there will be 6 instances of cb1 and one of cb2. f.add_done_callback(cb2) for i in range(5): f.add_done_callback(cb1) # Remove all instances of cb1. One cb2 remains. self.assertEqual(f.remove_done_callback(cb1), 6) self.assertEqual(bag, []) f.set_result('foo') self.run_briefly() self.assertEqual(bag, [2]) self.assertEqual(f.result(), 'foo') def test_remove_done_callbacks_list_mutation(self): # see http://bugs.python.org/issue28963 for details fut = self._new_future() fut.add_done_callback(str) for _ in range(63): fut.add_done_callback(id) class evil: def __eq__(self, other): fut.remove_done_callback(id) return False fut.remove_done_callback(evil()) def test_schedule_callbacks_list_mutation_1(self): # see http://bugs.python.org/issue28963 for details def mut(f): f.remove_done_callback(str) fut = self._new_future() fut.add_done_callback(mut) fut.add_done_callback(str) fut.add_done_callback(str) fut.set_result(1) test_utils.run_briefly(self.loop) def test_schedule_callbacks_list_mutation_2(self): # see http://bugs.python.org/issue30828 for details fut = self._new_future() fut.add_done_callback(str) for _ in range(63): fut.add_done_callback(id) max_extra_cbs = 100 extra_cbs = 0 class evil: def __eq__(self, other): nonlocal extra_cbs extra_cbs += 1 if extra_cbs < max_extra_cbs: fut.add_done_callback(id) return False fut.remove_done_callback(evil()) @unittest.skipUnless(hasattr(futures, '_CFuture'), 'requires the C _asyncio module') class CFutureDoneCallbackTests(BaseFutureDoneCallbackTests, test_utils.TestCase): def _new_future(self): return futures._CFuture(loop=self.loop) @unittest.skipUnless(hasattr(futures, '_CFuture'), 'requires the C _asyncio module') class CSubFutureDoneCallbackTests(BaseFutureDoneCallbackTests, test_utils.TestCase): def _new_future(self): class CSubFuture(futures._CFuture): pass return CSubFuture(loop=self.loop) class PyFutureDoneCallbackTests(BaseFutureDoneCallbackTests, test_utils.TestCase): def _new_future(self): return futures._PyFuture(loop=self.loop) if __name__ == '__main__': unittest.main()