| import threading | |
| import time | |
| import unittest | |
| from concurrent import futures | |
| from concurrent.futures._base import ( | |
| PENDING, RUNNING, CANCELLED, CANCELLED_AND_NOTIFIED, FINISHED, Future) | |
| from test import support | |
| from test.support import threading_helper | |
| from .util import ( | |
| PENDING_FUTURE, RUNNING_FUTURE, CANCELLED_FUTURE, | |
| CANCELLED_AND_NOTIFIED_FUTURE, EXCEPTION_FUTURE, SUCCESSFUL_FUTURE, | |
| BaseTestCase, create_future, setup_module) | |
| class FutureTests(BaseTestCase): | |
| def test_done_callback_with_result(self): | |
| callback_result = None | |
| def fn(callback_future): | |
| nonlocal callback_result | |
| callback_result = callback_future.result() | |
| f = Future() | |
| f.add_done_callback(fn) | |
| f.set_result(5) | |
| self.assertEqual(5, callback_result) | |
| def test_done_callback_with_exception(self): | |
| callback_exception = None | |
| def fn(callback_future): | |
| nonlocal callback_exception | |
| callback_exception = callback_future.exception() | |
| f = Future() | |
| f.add_done_callback(fn) | |
| f.set_exception(Exception('test')) | |
| self.assertEqual(('test',), callback_exception.args) | |
| def test_done_callback_with_cancel(self): | |
| was_cancelled = None | |
| def fn(callback_future): | |
| nonlocal was_cancelled | |
| was_cancelled = callback_future.cancelled() | |
| f = Future() | |
| f.add_done_callback(fn) | |
| self.assertTrue(f.cancel()) | |
| self.assertTrue(was_cancelled) | |
| def test_done_callback_raises(self): | |
| with support.captured_stderr() as stderr: | |
| raising_was_called = False | |
| fn_was_called = False | |
| def raising_fn(callback_future): | |
| nonlocal raising_was_called | |
| raising_was_called = True | |
| raise Exception('doh!') | |
| def fn(callback_future): | |
| nonlocal fn_was_called | |
| fn_was_called = True | |
| f = Future() | |
| f.add_done_callback(raising_fn) | |
| f.add_done_callback(fn) | |
| f.set_result(5) | |
| self.assertTrue(raising_was_called) | |
| self.assertTrue(fn_was_called) | |
| self.assertIn('Exception: doh!', stderr.getvalue()) | |
| def test_done_callback_already_successful(self): | |
| callback_result = None | |
| def fn(callback_future): | |
| nonlocal callback_result | |
| callback_result = callback_future.result() | |
| f = Future() | |
| f.set_result(5) | |
| f.add_done_callback(fn) | |
| self.assertEqual(5, callback_result) | |
| def test_done_callback_already_failed(self): | |
| callback_exception = None | |
| def fn(callback_future): | |
| nonlocal callback_exception | |
| callback_exception = callback_future.exception() | |
| f = Future() | |
| f.set_exception(Exception('test')) | |
| f.add_done_callback(fn) | |
| self.assertEqual(('test',), callback_exception.args) | |
| def test_done_callback_already_cancelled(self): | |
| was_cancelled = None | |
| def fn(callback_future): | |
| nonlocal was_cancelled | |
| was_cancelled = callback_future.cancelled() | |
| f = Future() | |
| self.assertTrue(f.cancel()) | |
| f.add_done_callback(fn) | |
| self.assertTrue(was_cancelled) | |
| def test_done_callback_raises_already_succeeded(self): | |
| with support.captured_stderr() as stderr: | |
| def raising_fn(callback_future): | |
| raise Exception('doh!') | |
| f = Future() | |
| # Set the result first to simulate a future that runs instantly, | |
| # effectively allowing the callback to be run immediately. | |
| f.set_result(5) | |
| f.add_done_callback(raising_fn) | |
| self.assertIn('exception calling callback for', stderr.getvalue()) | |
| self.assertIn('doh!', stderr.getvalue()) | |
| def test_repr(self): | |
| self.assertRegex(repr(PENDING_FUTURE), | |
| '<Future at 0x[0-9a-f]+ state=pending>') | |
| self.assertRegex(repr(RUNNING_FUTURE), | |
| '<Future at 0x[0-9a-f]+ state=running>') | |
| self.assertRegex(repr(CANCELLED_FUTURE), | |
| '<Future at 0x[0-9a-f]+ state=cancelled>') | |
| self.assertRegex(repr(CANCELLED_AND_NOTIFIED_FUTURE), | |
| '<Future at 0x[0-9a-f]+ state=cancelled>') | |
| self.assertRegex( | |
| repr(EXCEPTION_FUTURE), | |
| '<Future at 0x[0-9a-f]+ state=finished raised OSError>') | |
| self.assertRegex( | |
| repr(SUCCESSFUL_FUTURE), | |
| '<Future at 0x[0-9a-f]+ state=finished returned int>') | |
| def test_cancel(self): | |
| f1 = create_future(state=PENDING) | |
| f2 = create_future(state=RUNNING) | |
| f3 = create_future(state=CANCELLED) | |
| f4 = create_future(state=CANCELLED_AND_NOTIFIED) | |
| f5 = create_future(state=FINISHED, exception=OSError()) | |
| f6 = create_future(state=FINISHED, result=5) | |
| self.assertTrue(f1.cancel()) | |
| self.assertEqual(f1._state, CANCELLED) | |
| self.assertFalse(f2.cancel()) | |
| self.assertEqual(f2._state, RUNNING) | |
| self.assertTrue(f3.cancel()) | |
| self.assertEqual(f3._state, CANCELLED) | |
| self.assertTrue(f4.cancel()) | |
| self.assertEqual(f4._state, CANCELLED_AND_NOTIFIED) | |
| self.assertFalse(f5.cancel()) | |
| self.assertEqual(f5._state, FINISHED) | |
| self.assertFalse(f6.cancel()) | |
| self.assertEqual(f6._state, FINISHED) | |
| def test_cancelled(self): | |
| self.assertFalse(PENDING_FUTURE.cancelled()) | |
| self.assertFalse(RUNNING_FUTURE.cancelled()) | |
| self.assertTrue(CANCELLED_FUTURE.cancelled()) | |
| self.assertTrue(CANCELLED_AND_NOTIFIED_FUTURE.cancelled()) | |
| self.assertFalse(EXCEPTION_FUTURE.cancelled()) | |
| self.assertFalse(SUCCESSFUL_FUTURE.cancelled()) | |
| def test_done(self): | |
| self.assertFalse(PENDING_FUTURE.done()) | |
| self.assertFalse(RUNNING_FUTURE.done()) | |
| self.assertTrue(CANCELLED_FUTURE.done()) | |
| self.assertTrue(CANCELLED_AND_NOTIFIED_FUTURE.done()) | |
| self.assertTrue(EXCEPTION_FUTURE.done()) | |
| self.assertTrue(SUCCESSFUL_FUTURE.done()) | |
| def test_running(self): | |
| self.assertFalse(PENDING_FUTURE.running()) | |
| self.assertTrue(RUNNING_FUTURE.running()) | |
| self.assertFalse(CANCELLED_FUTURE.running()) | |
| self.assertFalse(CANCELLED_AND_NOTIFIED_FUTURE.running()) | |
| self.assertFalse(EXCEPTION_FUTURE.running()) | |
| self.assertFalse(SUCCESSFUL_FUTURE.running()) | |
| def test_result_with_timeout(self): | |
| self.assertRaises(futures.TimeoutError, | |
| PENDING_FUTURE.result, timeout=0) | |
| self.assertRaises(futures.TimeoutError, | |
| RUNNING_FUTURE.result, timeout=0) | |
| self.assertRaises(futures.CancelledError, | |
| CANCELLED_FUTURE.result, timeout=0) | |
| self.assertRaises(futures.CancelledError, | |
| CANCELLED_AND_NOTIFIED_FUTURE.result, timeout=0) | |
| self.assertRaises(OSError, EXCEPTION_FUTURE.result, timeout=0) | |
| self.assertEqual(SUCCESSFUL_FUTURE.result(timeout=0), 42) | |
| def test_result_with_success(self): | |
| # TODO(brian@sweetapp.com): This test is timing dependent. | |
| def notification(): | |
| # Wait until the main thread is waiting for the result. | |
| time.sleep(1) | |
| f1.set_result(42) | |
| f1 = create_future(state=PENDING) | |
| t = threading.Thread(target=notification) | |
| t.start() | |
| self.assertEqual(f1.result(timeout=5), 42) | |
| t.join() | |
| def test_result_with_cancel(self): | |
| # TODO(brian@sweetapp.com): This test is timing dependent. | |
| def notification(): | |
| # Wait until the main thread is waiting for the result. | |
| time.sleep(1) | |
| f1.cancel() | |
| f1 = create_future(state=PENDING) | |
| t = threading.Thread(target=notification) | |
| t.start() | |
| self.assertRaises(futures.CancelledError, | |
| f1.result, timeout=support.SHORT_TIMEOUT) | |
| t.join() | |
| def test_exception_with_timeout(self): | |
| self.assertRaises(futures.TimeoutError, | |
| PENDING_FUTURE.exception, timeout=0) | |
| self.assertRaises(futures.TimeoutError, | |
| RUNNING_FUTURE.exception, timeout=0) | |
| self.assertRaises(futures.CancelledError, | |
| CANCELLED_FUTURE.exception, timeout=0) | |
| self.assertRaises(futures.CancelledError, | |
| CANCELLED_AND_NOTIFIED_FUTURE.exception, timeout=0) | |
| self.assertTrue(isinstance(EXCEPTION_FUTURE.exception(timeout=0), | |
| OSError)) | |
| self.assertEqual(SUCCESSFUL_FUTURE.exception(timeout=0), None) | |
| def test_exception_with_success(self): | |
| def notification(): | |
| # Wait until the main thread is waiting for the exception. | |
| time.sleep(1) | |
| with f1._condition: | |
| f1._state = FINISHED | |
| f1._exception = OSError() | |
| f1._condition.notify_all() | |
| f1 = create_future(state=PENDING) | |
| t = threading.Thread(target=notification) | |
| t.start() | |
| self.assertTrue(isinstance(f1.exception(timeout=support.SHORT_TIMEOUT), OSError)) | |
| t.join() | |
| def test_multiple_set_result(self): | |
| f = create_future(state=PENDING) | |
| f.set_result(1) | |
| with self.assertRaisesRegex( | |
| futures.InvalidStateError, | |
| 'FINISHED: <Future at 0x[0-9a-f]+ ' | |
| 'state=finished returned int>' | |
| ): | |
| f.set_result(2) | |
| self.assertTrue(f.done()) | |
| self.assertEqual(f.result(), 1) | |
| def test_multiple_set_exception(self): | |
| f = create_future(state=PENDING) | |
| e = ValueError() | |
| f.set_exception(e) | |
| with self.assertRaisesRegex( | |
| futures.InvalidStateError, | |
| 'FINISHED: <Future at 0x[0-9a-f]+ ' | |
| 'state=finished raised ValueError>' | |
| ): | |
| f.set_exception(Exception()) | |
| self.assertEqual(f.exception(), e) | |
| def test_get_snapshot(self): | |
| """Test the _get_snapshot method for atomic state retrieval.""" | |
| # Test with a pending future | |
| f = Future() | |
| done, cancelled, result, exception = f._get_snapshot() | |
| self.assertFalse(done) | |
| self.assertFalse(cancelled) | |
| self.assertIsNone(result) | |
| self.assertIsNone(exception) | |
| # Test with a finished future (successful result) | |
| f = Future() | |
| f.set_result(42) | |
| done, cancelled, result, exception = f._get_snapshot() | |
| self.assertTrue(done) | |
| self.assertFalse(cancelled) | |
| self.assertEqual(result, 42) | |
| self.assertIsNone(exception) | |
| # Test with a finished future (exception) | |
| f = Future() | |
| exc = ValueError("test error") | |
| f.set_exception(exc) | |
| done, cancelled, result, exception = f._get_snapshot() | |
| self.assertTrue(done) | |
| self.assertFalse(cancelled) | |
| self.assertIsNone(result) | |
| self.assertIs(exception, exc) | |
| # Test with a cancelled future | |
| f = Future() | |
| f.cancel() | |
| done, cancelled, result, exception = f._get_snapshot() | |
| self.assertTrue(done) | |
| self.assertTrue(cancelled) | |
| self.assertIsNone(result) | |
| self.assertIsNone(exception) | |
| # Test concurrent access (basic thread safety check) | |
| f = Future() | |
| f.set_result(100) | |
| results = [] | |
| def get_snapshot(): | |
| for _ in range(1000): | |
| snapshot = f._get_snapshot() | |
| results.append(snapshot) | |
| threads = [threading.Thread(target=get_snapshot) for _ in range(4)] | |
| with threading_helper.start_threads(threads): | |
| pass | |
| # All snapshots should be identical for a finished future | |
| expected = (True, False, 100, None) | |
| for result in results: | |
| self.assertEqual(result, expected) | |
| def setUpModule(): | |
| setup_module() | |
| if __name__ == "__main__": | |
| unittest.main() |