Method test_copy_state

Lib/test/test_asyncio/test_futures.py:413–455

(self)

                         f'<{self.cls.__name__} cancelled>')

    def test_copy_state(self):
        from asyncio.futures import _copy_future_state

        f = concurrent.futures.Future()
        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 = concurrent.futures.Future()
        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 = concurrent.futures.Future()
        f_cancelled.cancel()

        newf_cancelled = self._new_future(loop=self.loop)
        _copy_future_state(f_cancelled, newf_cancelled)
        self.assertTrue(newf_cancelled.cancelled())

        try:
            raise concurrent.futures.InvalidStateError
        except BaseException as e:
            f_exc = e

        f_conexc = concurrent.futures.Future()
        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()
        except BaseException as e:
            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)

    def test_copy_state_from_concurrent_futures(self):
        """Test _copy_future_state from concurrent.futures.Future.