test_futures.py

"""Tests for futures.py."""

importconcurrent.futures
importgc
importre
importsys
importthreading
importtraceback
importunittest
fromunittestimportmock
fromtypesimportGenericAlias
importasyncio
fromasyncioimportfutures
importwarnings
fromtest.test_asyncioimportutilsastest_utils
fromtestimportsupport


deftearDownModule():
asyncio.set_event_loop_policy(None)


def_fakefunc(f):
returnf


deffirst_cb():
pass


deflast_cb():
pass


classReachableCode(Exception):
"""Exception to raise to indicate that some code was reached.

 Use this exception if using mocks is not a good alternative.
 """


classSimpleEvilEventLoop(asyncio.base_events.BaseEventLoop):
"""Base class for UAF and other evil stuff requiring an evil event loop."""

defget_debug(self): # to suppress tracebacks
returnFalse

def__del__(self):
# Automatically close the evil event loop to avoid warnings.
ifnotself.is_closed() andnotself.is_running():
self.close()


classDuckFuture:
# 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

defcancel(self):
ifself.done():
returnFalse
self.__cancelled=True
returnTrue

defcancelled(self):
returnself.__cancelled

defdone(self):
return (self.__cancelled
orself.__resultisnotNone
orself.__exceptionisnotNone)

defresult(self):
self.assertFalse(self.cancelled())
ifself.__exceptionisnotNone:
raiseself.__exception
returnself.__result

defexception(self):
self.assertFalse(self.cancelled())
returnself.__exception

defset_result(self, result):
self.assertFalse(self.done())
self.assertIsNotNone(result)
self.__result=result

defset_exception(self, exception):
self.assertFalse(self.done())
self.assertIsNotNone(exception)
self.__exception=exception

def__iter__(self):
ifnotself.done():
self._asyncio_future_blocking=True
yieldself
self.assertTrue(self.done())
returnself.result()


classDuckTests(test_utils.TestCase):

defsetUp(self):
super().setUp()
self.loop=self.new_test_loop()
self.addCleanup(self.loop.close)

deftest_wrap_future(self):
f=DuckFuture()
g=asyncio.wrap_future(f)
self.assertIs(g, f)

deftest_ensure_future(self):
f=DuckFuture()
g=asyncio.ensure_future(f)
self.assertIs(g, f)


classBaseFutureTests:

def_new_future(self, *args, **kwargs):
returnself.cls(*args, **kwargs)

defsetUp(self):
super().setUp()
self.loop=self.new_test_loop()
self.addCleanup(self.loop.close)

deftest_generic_alias(self):
future=self.cls[str]
self.assertEqual(future.__args__, (str,))
self.assertIsInstance(future, GenericAlias)

deftest_isfuture(self):
classMyFuture:
_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()

deftest_initial_state(self):
f=self._new_future(loop=self.loop)
self.assertFalse(f.cancelled())
self.assertFalse(f.done())
f.cancel()
self.assertTrue(f.cancelled())

deftest_constructor_without_loop(self):
withself.assertRaisesRegex(RuntimeError, 'no current event loop'):
self._new_future()

deftest_constructor_use_running_loop(self):
asyncdeftest():
returnself._new_future()
f=self.loop.run_until_complete(test())
self.assertIs(f._loop, self.loop)
self.assertIs(f.get_loop(), self.loop)

deftest_constructor_use_global_loop(self):
# Deprecated in 3.10, undeprecated in 3.12
asyncio.set_event_loop(self.loop)
self.addCleanup(asyncio.set_event_loop, None)
f=self._new_future()
self.assertIs(f._loop, self.loop)
self.assertIs(f.get_loop(), self.loop)

deftest_constructor_positional(self):
# Make sure Future doesn't accept a positional argument
self.assertRaises(TypeError, self._new_future, 42)

deftest_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)
withself.assertRaises((RuntimeError, AttributeError)):
fut.set_result(None)

fut=self.cls.__new__(self.cls, loop=self.loop)
withself.assertRaises((RuntimeError, AttributeError)):
fut.set_exception(Exception)

fut=self.cls.__new__(self.cls, loop=self.loop)
withself.assertRaises((RuntimeError, AttributeError)):
fut.cancel()

fut=self.cls.__new__(self.cls, loop=self.loop)
withself.assertRaises((RuntimeError, AttributeError)):
fut.add_done_callback(lambdaf: None)

fut=self.cls.__new__(self.cls, loop=self.loop)
withself.assertRaises((RuntimeError, AttributeError)):
fut.remove_done_callback(lambdaf: 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__()
exceptRuntimeError:
pass

fut=self.cls.__new__(self.cls, loop=self.loop)
try:
iter(fut)
exceptRuntimeError:
pass

fut=self.cls.__new__(self.cls, loop=self.loop)
self.assertFalse(fut.cancelled())
self.assertFalse(fut.done())

deftest_future_cancel_message_getter(self):
f=self._new_future(loop=self.loop)
self.assertTrue(hasattr(f, '_cancel_message'))
self.assertEqual(f._cancel_message, None)

f.cancel('my message')
withself.assertRaises(asyncio.CancelledError):
self.loop.run_until_complete(f)
self.assertEqual(f._cancel_message, 'my message')

deftest_future_cancel_message_setter(self):
f=self._new_future(loop=self.loop)
f.cancel('my message')
f._cancel_message='my new message'
self.assertEqual(f._cancel_message, 'my new message')

# Also check that the value is used for cancel().
withself.assertRaises(asyncio.CancelledError):
self.loop.run_until_complete(f)
self.assertEqual(f._cancel_message, 'my new message')

deftest_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())

deftest_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())

deftest_exception(self):
exc=RuntimeError()
f=self._new_future(loop=self.loop)
self.assertRaises(asyncio.InvalidStateError, f.exception)

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())

deftest_stop_iteration_exception(self, stop_iteration_class=StopIteration):
exc=stop_iteration_class()
f=self._new_future(loop=self.loop)
f.set_exception(exc)
self.assertFalse(f.cancelled())
self.assertTrue(f.done())
self.assertRaises(RuntimeError, f.result)
exc=f.exception()
cause=exc.__cause__
self.assertIsInstance(exc, RuntimeError)
self.assertRegex(str(exc), 'StopIteration .* cannot be raised')
self.assertIsInstance(cause, stop_iteration_class)

deftest_stop_iteration_subclass_exception(self):
classMyStopIteration(StopIteration):
pass

self.test_stop_iteration_exception(MyStopIteration)

deftest_exception_class(self):
f=self._new_future(loop=self.loop)
f.set_exception(RuntimeError)
self.assertIsInstance(f.exception(), RuntimeError)

deftest_yield_from_twice(self):
f=self._new_future(loop=self.loop)

deffixture():
yield'A'
x=yieldfromf
yield'B', x
y=yieldfromf
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.

deftest_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)

deffunc_repr(func):
filename, lineno=test_utils.get_function_source(func)
text='%s() at %s:%s'% (func.__qualname__, filename, lineno)
returnre.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)
foriinrange(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>')

deftest_copy_state(self):
fromasyncio.futuresimport_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())

try:
raiseconcurrent.futures.InvalidStateError
exceptBaseExceptionase:
f_exc=e

f_conexc=self._new_future(loop=self.loop)
f_conexc.set_exception(f_exc)

newf_conexc=self._new_future(loop=self.loop)
_copy_future_state(f_conexc, newf_conexc)
self.assertTrue(newf_conexc.done())
try:
newf_conexc.result()
exceptBaseExceptionase:
newf_exc=e# assertRaises context manager drops the traceback
newf_tb=''.join(traceback.format_tb(newf_exc.__traceback__))
self.assertEqual(newf_tb.count('raise concurrent.futures.InvalidStateError'), 1)

deftest_iter(self):
fut=self._new_future(loop=self.loop)

defcoro():
yieldfromfut

deftest():
arg1, arg2=coro()

withself.assertRaisesRegex(RuntimeError, "await wasn't used"):
test()
fut.cancel()

deftest_log_traceback(self):
fut=self._new_future(loop=self.loop)
withself.assertRaisesRegex(ValueError, 'can only be set to False'):
fut._log_traceback=True

@mock.patch('asyncio.base_events.logger')
deftest_tb_logger_abandoned(self, m_log):
fut=self._new_future(loop=self.loop)
delfut
self.assertFalse(m_log.error.called)

@mock.patch('asyncio.base_events.logger')
deftest_tb_logger_not_called_after_cancel(self, m_log):
fut=self._new_future(loop=self.loop)
fut.set_exception(Exception())
fut.cancel()
delfut
self.assertFalse(m_log.error.called)

@mock.patch('asyncio.base_events.logger')
deftest_tb_logger_result_unretrieved(self, m_log):
fut=self._new_future(loop=self.loop)
fut.set_result(42)
delfut
self.assertFalse(m_log.error.called)

@mock.patch('asyncio.base_events.logger')
deftest_tb_logger_result_retrieved(self, m_log):
fut=self._new_future(loop=self.loop)
fut.set_result(42)
fut.result()
delfut
self.assertFalse(m_log.error.called)

@mock.patch('asyncio.base_events.logger')
deftest_tb_logger_exception_unretrieved(self, m_log):
fut=self._new_future(loop=self.loop)
fut.set_exception(RuntimeError('boom'))
delfut
test_utils.run_briefly(self.loop)
support.gc_collect()
self.assertTrue(m_log.error.called)

@mock.patch('asyncio.base_events.logger')
deftest_tb_logger_exception_retrieved(self, m_log):
fut=self._new_future(loop=self.loop)
fut.set_exception(RuntimeError('boom'))
fut.exception()
delfut
self.assertFalse(m_log.error.called)

@mock.patch('asyncio.base_events.logger')
deftest_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)
delfut
self.assertFalse(m_log.error.called)

deftest_wrap_future(self):

defrun(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)

deftest_wrap_future_future(self):
f1=self._new_future(loop=self.loop)
f2=asyncio.wrap_future(f1)
self.assertIs(f1, f2)

deftest_wrap_future_without_loop(self):
defrun(arg):
return (arg, threading.get_ident())
ex=concurrent.futures.ThreadPoolExecutor(1)
f1=ex.submit(run, 'oi')
withself.assertRaisesRegex(RuntimeError, 'no current event loop'):
asyncio.wrap_future(f1)
ex.shutdown(wait=True)

deftest_wrap_future_use_running_loop(self):
defrun(arg):
return (arg, threading.get_ident())
ex=concurrent.futures.ThreadPoolExecutor(1)
f1=ex.submit(run, 'oi')
asyncdeftest():
returnasyncio.wrap_future(f1)
f2=self.loop.run_until_complete(test())
self.assertIs(self.loop, f2._loop)
ex.shutdown(wait=True)

deftest_wrap_future_use_global_loop(self):
# Deprecated in 3.10, undeprecated in 3.12
asyncio.set_event_loop(self.loop)
self.addCleanup(asyncio.set_event_loop, None)
defrun(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)

deftest_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())

deftest_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())

deftest_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')
defcheck_future_exception_never_retrieved(self, debug, m_log):
self.loop.set_debug(debug)

defmemory_error():
try:
raiseMemoryError()
exceptBaseExceptionasexc:
returnexc
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()

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)

message=m_log.error.call_args[0][0]
self.assertRegex(message, re.compile(regex, re.DOTALL))

deftest_future_exception_never_retrieved(self):
self.check_future_exception_never_retrieved(False)

deftest_future_exception_never_retrieved_debug(self):
self.check_future_exception_never_retrieved(True)

deftest_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())

deftest_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)
exceptStopIterationasex:
result=ex.args[0]
else:
self.fail('StopIteration was expected')
self.assertEqual(result, (1, 2))

deftest_future_iter_throw(self):
fut=self._new_future(loop=self.loop)
fi=iter(fut)
withself.assertWarns(DeprecationWarning):
self.assertRaises(Exception, fi.throw, Exception, Exception("zebra"), None)
withwarnings.catch_warnings():
warnings.filterwarnings("ignore", category=DeprecationWarning)
self.assertRaises(TypeError, fi.throw,
Exception, Exception("elephant"), 32)
self.assertRaises(TypeError, fi.throw,
Exception("elephant"), Exception("elephant"))
# https://github.com/python/cpython/issues/101326
self.assertRaises(ValueError, fi.throw, ValueError, None, None)
self.assertRaises(TypeError, fi.throw, list)

deftest_future_del_collect(self):
classEvil:
def__del__(self):
gc.collect()

foriinrange(100):
fut=self._new_future(loop=self.loop)
fut.set_result(Evil())

deftest_future_cancelled_result_refcycles(self):
f=self._new_future(loop=self.loop)
f.cancel()
exc=None
try:
f.result()
exceptasyncio.CancelledErrorase:
exc=e
self.assertIsNotNone(exc)
self.assertListEqual(gc.get_referrers(exc), [])

deftest_future_cancelled_exception_refcycles(self):
f=self._new_future(loop=self.loop)
f.cancel()
exc=None
try:
f.exception()
exceptasyncio.CancelledErrorase:
exc=e
self.assertIsNotNone(exc)
self.assertListEqual(gc.get_referrers(exc), [])


@unittest.skipUnless(hasattr(futures, '_CFuture'),
'requires the C _asyncio module')
classCFutureTests(BaseFutureTests, test_utils.TestCase):
try:
cls=futures._CFuture
exceptAttributeError:
cls=None

deftest_future_del_segfault(self):
fut=self._new_future(loop=self.loop)
withself.assertRaises(AttributeError):
delfut._asyncio_future_blocking
withself.assertRaises(AttributeError):
delfut._log_traceback

deftest_future_iter_get_referents_segfault(self):
# See https://github.com/python/cpython/issues/122695
import_asyncio
it=iter(self._new_future(loop=self.loop))
delit
evil=gc.get_referents(_asyncio)
gc.collect()

deftest_callbacks_copy(self):
# See https://github.com/python/cpython/issues/125789
# In C implementation, the `_callbacks` attribute
# always returns a new list to avoid mutations of internal state

fut=self._new_future(loop=self.loop)
f1=lambda_: 1
f2=lambda_: 2
fut.add_done_callback(f1)
fut.add_done_callback(f2)
callbacks=fut._callbacks
self.assertIsNot(callbacks, fut._callbacks)
fut.remove_done_callback(f1)
callbacks=fut._callbacks
self.assertIsNot(callbacks, fut._callbacks)
fut.remove_done_callback(f2)
self.assertIsNone(fut._callbacks)


@unittest.skipUnless(hasattr(futures, '_CFuture'),
'requires the C _asyncio module')
classCSubFutureTests(BaseFutureTests, test_utils.TestCase):
try:
classCSubFuture(futures._CFuture):
pass

cls=CSubFuture
exceptAttributeError:
cls=None


classPyFutureTests(BaseFutureTests, test_utils.TestCase):
cls=futures._PyFuture


classBaseFutureDoneCallbackTests():

defsetUp(self):
super().setUp()
self.loop=self.new_test_loop()

defrun_briefly(self):
test_utils.run_briefly(self.loop)

def_make_callback(self, bag, thing):
# Create a callback function that appends thing to bag.
defbag_appender(future):
bag.append(thing)
returnbag_appender

def_new_future(self):
raiseNotImplementedError

deftest_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')

deftest_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')

deftest_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')

deftest_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')

deftest_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)

deftest_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)
foriinrange(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')

deftest_remove_done_callbacks_list_mutation(self):
# see http://bugs.python.org/issue28963 for details

fut=self._new_future()
fut.add_done_callback(str)

for_inrange(63):
fut.add_done_callback(id)

classevil:
def__eq__(self, other):
fut.remove_done_callback(id)
returnFalse

fut.remove_done_callback(evil())

deftest_remove_done_callbacks_list_clear(self):
# see https://github.com/python/cpython/issues/97592 for details

fut=self._new_future()
fut.add_done_callback(str)

for_inrange(63):
fut.add_done_callback(id)

classevil:
def__eq__(self, other):
fut.remove_done_callback(other)

fut.remove_done_callback(evil())

deftest_schedule_callbacks_list_mutation_1(self):
# see http://bugs.python.org/issue28963 for details

defmut(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)

deftest_schedule_callbacks_list_mutation_2(self):
# see http://bugs.python.org/issue30828 for details

fut=self._new_future()
fut.add_done_callback(str)

for_inrange(63):
fut.add_done_callback(id)

max_extra_cbs=100
extra_cbs=0

classevil:
def__eq__(self, other):
nonlocalextra_cbs
extra_cbs+=1
ifextra_cbs<max_extra_cbs:
fut.add_done_callback(id)
returnFalse

fut.remove_done_callback(evil())

deftest_evil_call_soon_list_mutation(self):
# see: https://github.com/python/cpython/issues/125969
called_on_fut_callback0=False

pad=lambda: ...

defevil_call_soon(*args, **kwargs):
nonlocalcalled_on_fut_callback0
ifcalled_on_fut_callback0:
# Called when handling fut->fut_callbacks[0]
# and mutates the length fut->fut_callbacks.
fut.remove_done_callback(int)
fut.remove_done_callback(pad)
else:
called_on_fut_callback0=True

fake_event_loop=SimpleEvilEventLoop()
fake_event_loop.call_soon=evil_call_soon

withmock.patch.object(self, 'loop', fake_event_loop):
fut=self._new_future()
self.assertIs(fut.get_loop(), fake_event_loop)

fut.add_done_callback(str) # sets fut->fut_callback0