Python code coverage for Lib/test/test_concurrent_futures.py

#	count	content
1	n/a	import test.support
2	n/a
3	n/a	# Skip tests if _multiprocessing wasn't built.
4	n/a	test.support.import_module('_multiprocessing')
5	n/a	# Skip tests if sem_open implementation is broken.
6	n/a	test.support.import_module('multiprocessing.synchronize')
7	n/a	# import threading after _multiprocessing to raise a more relevant error
8	n/a	# message: "No module named _multiprocessing". _multiprocessing is not compiled
9	n/a	# without thread support.
10	n/a	test.support.import_module('threading')
11	n/a
12	n/a	from test.support.script_helper import assert_python_ok
13	n/a
14	n/a	import os
15	n/a	import sys
16	n/a	import threading
17	n/a	import time
18	n/a	import unittest
19	n/a	import weakref
20	n/a
21	n/a	from concurrent import futures
22	n/a	from concurrent.futures._base import (
23	n/a	PENDING, RUNNING, CANCELLED, CANCELLED_AND_NOTIFIED, FINISHED, Future)
24	n/a	from concurrent.futures.process import BrokenProcessPool
25	n/a
26	n/a
27	n/a	def create_future(state=PENDING, exception=None, result=None):
28	n/a	f = Future()
29	n/a	f._state = state
30	n/a	f._exception = exception
31	n/a	f._result = result
32	n/a	return f
33	n/a
34	n/a
35	n/a	PENDING_FUTURE = create_future(state=PENDING)
36	n/a	RUNNING_FUTURE = create_future(state=RUNNING)
37	n/a	CANCELLED_FUTURE = create_future(state=CANCELLED)
38	n/a	CANCELLED_AND_NOTIFIED_FUTURE = create_future(state=CANCELLED_AND_NOTIFIED)
39	n/a	EXCEPTION_FUTURE = create_future(state=FINISHED, exception=OSError())
40	n/a	SUCCESSFUL_FUTURE = create_future(state=FINISHED, result=42)
41	n/a
42	n/a
43	n/a	def mul(x, y):
44	n/a	return x * y
45	n/a
46	n/a
47	n/a	def sleep_and_raise(t):
48	n/a	time.sleep(t)
49	n/a	raise Exception('this is an exception')
50	n/a
51	n/a	def sleep_and_print(t, msg):
52	n/a	time.sleep(t)
53	n/a	print(msg)
54	n/a	sys.stdout.flush()
55	n/a
56	n/a
57	n/a	class MyObject(object):
58	n/a	def my_method(self):
59	n/a	pass
60	n/a
61	n/a
62	n/a	class ExecutorMixin:
63	n/a	worker_count = 5
64	n/a
65	n/a	def setUp(self):
66	n/a	self.t1 = time.time()
67	n/a	try:
68	n/a	self.executor = self.executor_type(max_workers=self.worker_count)
69	n/a	except NotImplementedError as e:
70	n/a	self.skipTest(str(e))
71	n/a	self._prime_executor()
72	n/a
73	n/a	def tearDown(self):
74	n/a	self.executor.shutdown(wait=True)
75	n/a	dt = time.time() - self.t1
76	n/a	if test.support.verbose:
77	n/a	print("%.2fs" % dt, end=' ')
78	n/a	self.assertLess(dt, 60, "synchronization issue: test lasted too long")
79	n/a
80	n/a	def _prime_executor(self):
81	n/a	# Make sure that the executor is ready to do work before running the
82	n/a	# tests. This should reduce the probability of timeouts in the tests.
83	n/a	futures = [self.executor.submit(time.sleep, 0.1)
84	n/a	for _ in range(self.worker_count)]
85	n/a
86	n/a	for f in futures:
87	n/a	f.result()
88	n/a
89	n/a
90	n/a	class ThreadPoolMixin(ExecutorMixin):
91	n/a	executor_type = futures.ThreadPoolExecutor
92	n/a
93	n/a
94	n/a	class ProcessPoolMixin(ExecutorMixin):
95	n/a	executor_type = futures.ProcessPoolExecutor
96	n/a
97	n/a
98	n/a	class ExecutorShutdownTest:
99	n/a	def test_run_after_shutdown(self):
100	n/a	self.executor.shutdown()
101	n/a	self.assertRaises(RuntimeError,
102	n/a	self.executor.submit,
103	n/a	pow, 2, 5)
104	n/a
105	n/a	def test_interpreter_shutdown(self):
106	n/a	# Test the atexit hook for shutdown of worker threads and processes
107	n/a	rc, out, err = assert_python_ok('-c', """if 1:
108	n/a	from concurrent.futures import {executor_type}
109	n/a	from time import sleep
110	n/a	from test.test_concurrent_futures import sleep_and_print
111	n/a	t = {executor_type}(5)
112	n/a	t.submit(sleep_and_print, 1.0, "apple")
113	n/a	""".format(executor_type=self.executor_type.__name__))
114	n/a	# Errors in atexit hooks don't change the process exit code, check
115	n/a	# stderr manually.
116	n/a	self.assertFalse(err)
117	n/a	self.assertEqual(out.strip(), b"apple")
118	n/a
119	n/a	def test_hang_issue12364(self):
120	n/a	fs = [self.executor.submit(time.sleep, 0.1) for _ in range(50)]
121	n/a	self.executor.shutdown()
122	n/a	for f in fs:
123	n/a	f.result()
124	n/a
125	n/a
126	n/a	class ThreadPoolShutdownTest(ThreadPoolMixin, ExecutorShutdownTest, unittest.TestCase):
127	n/a	def _prime_executor(self):
128	n/a	pass
129	n/a
130	n/a	def test_threads_terminate(self):
131	n/a	self.executor.submit(mul, 21, 2)
132	n/a	self.executor.submit(mul, 6, 7)
133	n/a	self.executor.submit(mul, 3, 14)
134	n/a	self.assertEqual(len(self.executor._threads), 3)
135	n/a	self.executor.shutdown()
136	n/a	for t in self.executor._threads:
137	n/a	t.join()
138	n/a
139	n/a	def test_context_manager_shutdown(self):
140	n/a	with futures.ThreadPoolExecutor(max_workers=5) as e:
141	n/a	executor = e
142	n/a	self.assertEqual(list(e.map(abs, range(-5, 5))),
143	n/a	[5, 4, 3, 2, 1, 0, 1, 2, 3, 4])
144	n/a
145	n/a	for t in executor._threads:
146	n/a	t.join()
147	n/a
148	n/a	def test_del_shutdown(self):
149	n/a	executor = futures.ThreadPoolExecutor(max_workers=5)
150	n/a	executor.map(abs, range(-5, 5))
151	n/a	threads = executor._threads
152	n/a	del executor
153	n/a
154	n/a	for t in threads:
155	n/a	t.join()
156	n/a
157	n/a	def test_thread_names_assigned(self):
158	n/a	executor = futures.ThreadPoolExecutor(
159	n/a	max_workers=5, thread_name_prefix='SpecialPool')
160	n/a	executor.map(abs, range(-5, 5))
161	n/a	threads = executor._threads
162	n/a	del executor
163	n/a
164	n/a	for t in threads:
165	n/a	self.assertRegex(t.name, r'^SpecialPool_[0-4]$')
166	n/a	t.join()
167	n/a
168	n/a	def test_thread_names_default(self):
169	n/a	executor = futures.ThreadPoolExecutor(max_workers=5)
170	n/a	executor.map(abs, range(-5, 5))
171	n/a	threads = executor._threads
172	n/a	del executor
173	n/a
174	n/a	for t in threads:
175	n/a	# We don't particularly care what the default name is, just that
176	n/a	# it has a default name implying that it is a ThreadPoolExecutor
177	n/a	# followed by what looks like a thread number.
178	n/a	self.assertRegex(t.name, r'^.ThreadPoolExecutor._[0-4]$')
179	n/a	t.join()
180	n/a
181	n/a
182	n/a	class ProcessPoolShutdownTest(ProcessPoolMixin, ExecutorShutdownTest, unittest.TestCase):
183	n/a	def _prime_executor(self):
184	n/a	pass
185	n/a
186	n/a	def test_processes_terminate(self):
187	n/a	self.executor.submit(mul, 21, 2)
188	n/a	self.executor.submit(mul, 6, 7)
189	n/a	self.executor.submit(mul, 3, 14)
190	n/a	self.assertEqual(len(self.executor._processes), 5)
191	n/a	processes = self.executor._processes
192	n/a	self.executor.shutdown()
193	n/a
194	n/a	for p in processes.values():
195	n/a	p.join()
196	n/a
197	n/a	def test_context_manager_shutdown(self):
198	n/a	with futures.ProcessPoolExecutor(max_workers=5) as e:
199	n/a	processes = e._processes
200	n/a	self.assertEqual(list(e.map(abs, range(-5, 5))),
201	n/a	[5, 4, 3, 2, 1, 0, 1, 2, 3, 4])
202	n/a
203	n/a	for p in processes.values():
204	n/a	p.join()
205	n/a
206	n/a	def test_del_shutdown(self):
207	n/a	executor = futures.ProcessPoolExecutor(max_workers=5)
208	n/a	list(executor.map(abs, range(-5, 5)))
209	n/a	queue_management_thread = executor._queue_management_thread
210	n/a	processes = executor._processes
211	n/a	del executor
212	n/a
213	n/a	queue_management_thread.join()
214	n/a	for p in processes.values():
215	n/a	p.join()
216	n/a
217	n/a
218	n/a	class WaitTests:
219	n/a
220	n/a	def test_first_completed(self):
221	n/a	future1 = self.executor.submit(mul, 21, 2)
222	n/a	future2 = self.executor.submit(time.sleep, 1.5)
223	n/a
224	n/a	done, not_done = futures.wait(
225	n/a	[CANCELLED_FUTURE, future1, future2],
226	n/a	return_when=futures.FIRST_COMPLETED)
227	n/a
228	n/a	self.assertEqual(set([future1]), done)
229	n/a	self.assertEqual(set([CANCELLED_FUTURE, future2]), not_done)
230	n/a
231	n/a	def test_first_completed_some_already_completed(self):
232	n/a	future1 = self.executor.submit(time.sleep, 1.5)
233	n/a
234	n/a	finished, pending = futures.wait(
235	n/a	[CANCELLED_AND_NOTIFIED_FUTURE, SUCCESSFUL_FUTURE, future1],
236	n/a	return_when=futures.FIRST_COMPLETED)
237	n/a
238	n/a	self.assertEqual(
239	n/a	set([CANCELLED_AND_NOTIFIED_FUTURE, SUCCESSFUL_FUTURE]),
240	n/a	finished)
241	n/a	self.assertEqual(set([future1]), pending)
242	n/a
243	n/a	def test_first_exception(self):
244	n/a	future1 = self.executor.submit(mul, 2, 21)
245	n/a	future2 = self.executor.submit(sleep_and_raise, 1.5)
246	n/a	future3 = self.executor.submit(time.sleep, 3)
247	n/a
248	n/a	finished, pending = futures.wait(
249	n/a	[future1, future2, future3],
250	n/a	return_when=futures.FIRST_EXCEPTION)
251	n/a
252	n/a	self.assertEqual(set([future1, future2]), finished)
253	n/a	self.assertEqual(set([future3]), pending)
254	n/a
255	n/a	def test_first_exception_some_already_complete(self):
256	n/a	future1 = self.executor.submit(divmod, 21, 0)
257	n/a	future2 = self.executor.submit(time.sleep, 1.5)
258	n/a
259	n/a	finished, pending = futures.wait(
260	n/a	[SUCCESSFUL_FUTURE,
261	n/a	CANCELLED_FUTURE,
262	n/a	CANCELLED_AND_NOTIFIED_FUTURE,
263	n/a	future1, future2],
264	n/a	return_when=futures.FIRST_EXCEPTION)
265	n/a
266	n/a	self.assertEqual(set([SUCCESSFUL_FUTURE,
267	n/a	CANCELLED_AND_NOTIFIED_FUTURE,
268	n/a	future1]), finished)
269	n/a	self.assertEqual(set([CANCELLED_FUTURE, future2]), pending)
270	n/a
271	n/a	def test_first_exception_one_already_failed(self):
272	n/a	future1 = self.executor.submit(time.sleep, 2)
273	n/a
274	n/a	finished, pending = futures.wait(
275	n/a	[EXCEPTION_FUTURE, future1],
276	n/a	return_when=futures.FIRST_EXCEPTION)
277	n/a
278	n/a	self.assertEqual(set([EXCEPTION_FUTURE]), finished)
279	n/a	self.assertEqual(set([future1]), pending)
280	n/a
281	n/a	def test_all_completed(self):
282	n/a	future1 = self.executor.submit(divmod, 2, 0)
283	n/a	future2 = self.executor.submit(mul, 2, 21)
284	n/a
285	n/a	finished, pending = futures.wait(
286	n/a	[SUCCESSFUL_FUTURE,
287	n/a	CANCELLED_AND_NOTIFIED_FUTURE,
288	n/a	EXCEPTION_FUTURE,
289	n/a	future1,
290	n/a	future2],
291	n/a	return_when=futures.ALL_COMPLETED)
292	n/a
293	n/a	self.assertEqual(set([SUCCESSFUL_FUTURE,
294	n/a	CANCELLED_AND_NOTIFIED_FUTURE,
295	n/a	EXCEPTION_FUTURE,
296	n/a	future1,
297	n/a	future2]), finished)
298	n/a	self.assertEqual(set(), pending)
299	n/a
300	n/a	def test_timeout(self):
301	n/a	future1 = self.executor.submit(mul, 6, 7)
302	n/a	future2 = self.executor.submit(time.sleep, 6)
303	n/a
304	n/a	finished, pending = futures.wait(
305	n/a	[CANCELLED_AND_NOTIFIED_FUTURE,
306	n/a	EXCEPTION_FUTURE,
307	n/a	SUCCESSFUL_FUTURE,
308	n/a	future1, future2],
309	n/a	timeout=5,
310	n/a	return_when=futures.ALL_COMPLETED)
311	n/a
312	n/a	self.assertEqual(set([CANCELLED_AND_NOTIFIED_FUTURE,
313	n/a	EXCEPTION_FUTURE,
314	n/a	SUCCESSFUL_FUTURE,
315	n/a	future1]), finished)
316	n/a	self.assertEqual(set([future2]), pending)
317	n/a
318	n/a
319	n/a	class ThreadPoolWaitTests(ThreadPoolMixin, WaitTests, unittest.TestCase):
320	n/a
321	n/a	def test_pending_calls_race(self):
322	n/a	# Issue #14406: multi-threaded race condition when waiting on all
323	n/a	# futures.
324	n/a	event = threading.Event()
325	n/a	def future_func():
326	n/a	event.wait()
327	n/a	oldswitchinterval = sys.getswitchinterval()
328	n/a	sys.setswitchinterval(1e-6)
329	n/a	try:
330	n/a	fs = {self.executor.submit(future_func) for i in range(100)}
331	n/a	event.set()
332	n/a	futures.wait(fs, return_when=futures.ALL_COMPLETED)
333	n/a	finally:
334	n/a	sys.setswitchinterval(oldswitchinterval)
335	n/a
336	n/a
337	n/a	class ProcessPoolWaitTests(ProcessPoolMixin, WaitTests, unittest.TestCase):
338	n/a	pass
339	n/a
340	n/a
341	n/a	class AsCompletedTests:
342	n/a	# TODO(brian@sweetapp.com): Should have a test with a non-zero timeout.
343	n/a	def test_no_timeout(self):
344	n/a	future1 = self.executor.submit(mul, 2, 21)
345	n/a	future2 = self.executor.submit(mul, 7, 6)
346	n/a
347	n/a	completed = set(futures.as_completed(
348	n/a	[CANCELLED_AND_NOTIFIED_FUTURE,
349	n/a	EXCEPTION_FUTURE,
350	n/a	SUCCESSFUL_FUTURE,
351	n/a	future1, future2]))
352	n/a	self.assertEqual(set(
353	n/a	[CANCELLED_AND_NOTIFIED_FUTURE,
354	n/a	EXCEPTION_FUTURE,
355	n/a	SUCCESSFUL_FUTURE,
356	n/a	future1, future2]),
357	n/a	completed)
358	n/a
359	n/a	def test_zero_timeout(self):
360	n/a	future1 = self.executor.submit(time.sleep, 2)
361	n/a	completed_futures = set()
362	n/a	try:
363	n/a	for future in futures.as_completed(
364	n/a	[CANCELLED_AND_NOTIFIED_FUTURE,
365	n/a	EXCEPTION_FUTURE,
366	n/a	SUCCESSFUL_FUTURE,
367	n/a	future1],
368	n/a	timeout=0):
369	n/a	completed_futures.add(future)
370	n/a	except futures.TimeoutError:
371	n/a	pass
372	n/a
373	n/a	self.assertEqual(set([CANCELLED_AND_NOTIFIED_FUTURE,
374	n/a	EXCEPTION_FUTURE,
375	n/a	SUCCESSFUL_FUTURE]),
376	n/a	completed_futures)
377	n/a
378	n/a	def test_duplicate_futures(self):
379	n/a	# Issue 20367. Duplicate futures should not raise exceptions or give
380	n/a	# duplicate responses.
381	n/a	future1 = self.executor.submit(time.sleep, 2)
382	n/a	completed = [f for f in futures.as_completed([future1,future1])]
383	n/a	self.assertEqual(len(completed), 1)
384	n/a
385	n/a
386	n/a	class ThreadPoolAsCompletedTests(ThreadPoolMixin, AsCompletedTests, unittest.TestCase):
387	n/a	pass
388	n/a
389	n/a
390	n/a	class ProcessPoolAsCompletedTests(ProcessPoolMixin, AsCompletedTests, unittest.TestCase):
391	n/a	pass
392	n/a
393	n/a
394	n/a	class ExecutorTest:
395	n/a	# Executor.shutdown() and context manager usage is tested by
396	n/a	# ExecutorShutdownTest.
397	n/a	def test_submit(self):
398	n/a	future = self.executor.submit(pow, 2, 8)
399	n/a	self.assertEqual(256, future.result())
400	n/a
401	n/a	def test_submit_keyword(self):
402	n/a	future = self.executor.submit(mul, 2, y=8)
403	n/a	self.assertEqual(16, future.result())
404	n/a
405	n/a	def test_map(self):
406	n/a	self.assertEqual(
407	n/a	list(self.executor.map(pow, range(10), range(10))),
408	n/a	list(map(pow, range(10), range(10))))
409	n/a
410	n/a	def test_map_exception(self):
411	n/a	i = self.executor.map(divmod, [1, 1, 1, 1], [2, 3, 0, 5])
412	n/a	self.assertEqual(i.__next__(), (0, 1))
413	n/a	self.assertEqual(i.__next__(), (0, 1))
414	n/a	self.assertRaises(ZeroDivisionError, i.__next__)
415	n/a
416	n/a	def test_map_timeout(self):
417	n/a	results = []
418	n/a	try:
419	n/a	for i in self.executor.map(time.sleep,
420	n/a	[0, 0, 6],
421	n/a	timeout=5):
422	n/a	results.append(i)
423	n/a	except futures.TimeoutError:
424	n/a	pass
425	n/a	else:
426	n/a	self.fail('expected TimeoutError')
427	n/a
428	n/a	self.assertEqual([None, None], results)
429	n/a
430	n/a	def test_shutdown_race_issue12456(self):
431	n/a	# Issue #12456: race condition at shutdown where trying to post a
432	n/a	# sentinel in the call queue blocks (the queue is full while processes
433	n/a	# have exited).
434	n/a	self.executor.map(str, [2] * (self.worker_count + 1))
435	n/a	self.executor.shutdown()
436	n/a
437	n/a	@test.support.cpython_only
438	n/a	def test_no_stale_references(self):
439	n/a	# Issue #16284: check that the executors don't unnecessarily hang onto
440	n/a	# references.
441	n/a	my_object = MyObject()
442	n/a	my_object_collected = threading.Event()
443	n/a	my_object_callback = weakref.ref(
444	n/a	my_object, lambda obj: my_object_collected.set())
445	n/a	# Deliberately discarding the future.
446	n/a	self.executor.submit(my_object.my_method)
447	n/a	del my_object
448	n/a
449	n/a	collected = my_object_collected.wait(timeout=5.0)
450	n/a	self.assertTrue(collected,
451	n/a	"Stale reference not collected within timeout.")
452	n/a
453	n/a	def test_max_workers_negative(self):
454	n/a	for number in (0, -1):
455	n/a	with self.assertRaisesRegex(ValueError,
456	n/a	"max_workers must be greater "
457	n/a	"than 0"):
458	n/a	self.executor_type(max_workers=number)
459	n/a
460	n/a
461	n/a	class ThreadPoolExecutorTest(ThreadPoolMixin, ExecutorTest, unittest.TestCase):
462	n/a	def test_map_submits_without_iteration(self):
463	n/a	"""Tests verifying issue 11777."""
464	n/a	finished = []
465	n/a	def record_finished(n):
466	n/a	finished.append(n)
467	n/a
468	n/a	self.executor.map(record_finished, range(10))
469	n/a	self.executor.shutdown(wait=True)
470	n/a	self.assertCountEqual(finished, range(10))
471	n/a
472	n/a	def test_default_workers(self):
473	n/a	executor = self.executor_type()
474	n/a	self.assertEqual(executor._max_workers,
475	n/a	(os.cpu_count() or 1) * 5)
476	n/a
477	n/a
478	n/a	class ProcessPoolExecutorTest(ProcessPoolMixin, ExecutorTest, unittest.TestCase):
479	n/a	def test_killed_child(self):
480	n/a	# When a child process is abruptly terminated, the whole pool gets
481	n/a	# "broken".
482	n/a	futures = [self.executor.submit(time.sleep, 3)]
483	n/a	# Get one of the processes, and terminate (kill) it
484	n/a	p = next(iter(self.executor._processes.values()))
485	n/a	p.terminate()
486	n/a	for fut in futures:
487	n/a	self.assertRaises(BrokenProcessPool, fut.result)
488	n/a	# Submitting other jobs fails as well.
489	n/a	self.assertRaises(BrokenProcessPool, self.executor.submit, pow, 2, 8)
490	n/a
491	n/a	def test_map_chunksize(self):
492	n/a	def bad_map():
493	n/a	list(self.executor.map(pow, range(40), range(40), chunksize=-1))
494	n/a
495	n/a	ref = list(map(pow, range(40), range(40)))
496	n/a	self.assertEqual(
497	n/a	list(self.executor.map(pow, range(40), range(40), chunksize=6)),
498	n/a	ref)
499	n/a	self.assertEqual(
500	n/a	list(self.executor.map(pow, range(40), range(40), chunksize=50)),
501	n/a	ref)
502	n/a	self.assertEqual(
503	n/a	list(self.executor.map(pow, range(40), range(40), chunksize=40)),
504	n/a	ref)
505	n/a	self.assertRaises(ValueError, bad_map)
506	n/a
507	n/a	@classmethod
508	n/a	def _test_traceback(cls):
509	n/a	raise RuntimeError(123) # some comment
510	n/a
511	n/a	def test_traceback(self):
512	n/a	# We want ensure that the traceback from the child process is
513	n/a	# contained in the traceback raised in the main process.
514	n/a	future = self.executor.submit(self._test_traceback)
515	n/a	with self.assertRaises(Exception) as cm:
516	n/a	future.result()
517	n/a
518	n/a	exc = cm.exception
519	n/a	self.assertIs(type(exc), RuntimeError)
520	n/a	self.assertEqual(exc.args, (123,))
521	n/a	cause = exc.__cause__
522	n/a	self.assertIs(type(cause), futures.process._RemoteTraceback)
523	n/a	self.assertIn('raise RuntimeError(123) # some comment', cause.tb)
524	n/a
525	n/a	with test.support.captured_stderr() as f1:
526	n/a	try:
527	n/a	raise exc
528	n/a	except RuntimeError:
529	n/a	sys.excepthook(*sys.exc_info())
530	n/a	self.assertIn('raise RuntimeError(123) # some comment',
531	n/a	f1.getvalue())
532	n/a
533	n/a
534	n/a	class FutureTests(unittest.TestCase):
535	n/a	def test_done_callback_with_result(self):
536	n/a	callback_result = None
537	n/a	def fn(callback_future):
538	n/a	nonlocal callback_result
539	n/a	callback_result = callback_future.result()
540	n/a
541	n/a	f = Future()
542	n/a	f.add_done_callback(fn)
543	n/a	f.set_result(5)
544	n/a	self.assertEqual(5, callback_result)
545	n/a
546	n/a	def test_done_callback_with_exception(self):
547	n/a	callback_exception = None
548	n/a	def fn(callback_future):
549	n/a	nonlocal callback_exception
550	n/a	callback_exception = callback_future.exception()
551	n/a
552	n/a	f = Future()
553	n/a	f.add_done_callback(fn)
554	n/a	f.set_exception(Exception('test'))
555	n/a	self.assertEqual(('test',), callback_exception.args)
556	n/a
557	n/a	def test_done_callback_with_cancel(self):
558	n/a	was_cancelled = None
559	n/a	def fn(callback_future):
560	n/a	nonlocal was_cancelled
561	n/a	was_cancelled = callback_future.cancelled()
562	n/a
563	n/a	f = Future()
564	n/a	f.add_done_callback(fn)
565	n/a	self.assertTrue(f.cancel())
566	n/a	self.assertTrue(was_cancelled)
567	n/a
568	n/a	def test_done_callback_raises(self):
569	n/a	with test.support.captured_stderr() as stderr:
570	n/a	raising_was_called = False
571	n/a	fn_was_called = False
572	n/a
573	n/a	def raising_fn(callback_future):
574	n/a	nonlocal raising_was_called
575	n/a	raising_was_called = True
576	n/a	raise Exception('doh!')
577	n/a
578	n/a	def fn(callback_future):
579	n/a	nonlocal fn_was_called
580	n/a	fn_was_called = True
581	n/a
582	n/a	f = Future()
583	n/a	f.add_done_callback(raising_fn)
584	n/a	f.add_done_callback(fn)
585	n/a	f.set_result(5)
586	n/a	self.assertTrue(raising_was_called)
587	n/a	self.assertTrue(fn_was_called)
588	n/a	self.assertIn('Exception: doh!', stderr.getvalue())
589	n/a
590	n/a	def test_done_callback_already_successful(self):
591	n/a	callback_result = None
592	n/a	def fn(callback_future):
593	n/a	nonlocal callback_result
594	n/a	callback_result = callback_future.result()
595	n/a
596	n/a	f = Future()
597	n/a	f.set_result(5)
598	n/a	f.add_done_callback(fn)
599	n/a	self.assertEqual(5, callback_result)
600	n/a
601	n/a	def test_done_callback_already_failed(self):
602	n/a	callback_exception = None
603	n/a	def fn(callback_future):
604	n/a	nonlocal callback_exception
605	n/a	callback_exception = callback_future.exception()
606	n/a
607	n/a	f = Future()
608	n/a	f.set_exception(Exception('test'))
609	n/a	f.add_done_callback(fn)
610	n/a	self.assertEqual(('test',), callback_exception.args)
611	n/a
612	n/a	def test_done_callback_already_cancelled(self):
613	n/a	was_cancelled = None
614	n/a	def fn(callback_future):
615	n/a	nonlocal was_cancelled
616	n/a	was_cancelled = callback_future.cancelled()
617	n/a
618	n/a	f = Future()
619	n/a	self.assertTrue(f.cancel())
620	n/a	f.add_done_callback(fn)
621	n/a	self.assertTrue(was_cancelled)
622	n/a
623	n/a	def test_repr(self):
624	n/a	self.assertRegex(repr(PENDING_FUTURE),
625	n/a	'<Future at 0x[0-9a-f]+ state=pending>')
626	n/a	self.assertRegex(repr(RUNNING_FUTURE),
627	n/a	'<Future at 0x[0-9a-f]+ state=running>')
628	n/a	self.assertRegex(repr(CANCELLED_FUTURE),
629	n/a	'<Future at 0x[0-9a-f]+ state=cancelled>')
630	n/a	self.assertRegex(repr(CANCELLED_AND_NOTIFIED_FUTURE),
631	n/a	'<Future at 0x[0-9a-f]+ state=cancelled>')
632	n/a	self.assertRegex(
633	n/a	repr(EXCEPTION_FUTURE),
634	n/a	'<Future at 0x[0-9a-f]+ state=finished raised OSError>')
635	n/a	self.assertRegex(
636	n/a	repr(SUCCESSFUL_FUTURE),
637	n/a	'<Future at 0x[0-9a-f]+ state=finished returned int>')
638	n/a
639	n/a
640	n/a	def test_cancel(self):
641	n/a	f1 = create_future(state=PENDING)
642	n/a	f2 = create_future(state=RUNNING)
643	n/a	f3 = create_future(state=CANCELLED)
644	n/a	f4 = create_future(state=CANCELLED_AND_NOTIFIED)
645	n/a	f5 = create_future(state=FINISHED, exception=OSError())
646	n/a	f6 = create_future(state=FINISHED, result=5)
647	n/a
648	n/a	self.assertTrue(f1.cancel())
649	n/a	self.assertEqual(f1._state, CANCELLED)
650	n/a
651	n/a	self.assertFalse(f2.cancel())
652	n/a	self.assertEqual(f2._state, RUNNING)
653	n/a
654	n/a	self.assertTrue(f3.cancel())
655	n/a	self.assertEqual(f3._state, CANCELLED)
656	n/a
657	n/a	self.assertTrue(f4.cancel())
658	n/a	self.assertEqual(f4._state, CANCELLED_AND_NOTIFIED)
659	n/a
660	n/a	self.assertFalse(f5.cancel())
661	n/a	self.assertEqual(f5._state, FINISHED)
662	n/a
663	n/a	self.assertFalse(f6.cancel())
664	n/a	self.assertEqual(f6._state, FINISHED)
665	n/a
666	n/a	def test_cancelled(self):
667	n/a	self.assertFalse(PENDING_FUTURE.cancelled())
668	n/a	self.assertFalse(RUNNING_FUTURE.cancelled())
669	n/a	self.assertTrue(CANCELLED_FUTURE.cancelled())
670	n/a	self.assertTrue(CANCELLED_AND_NOTIFIED_FUTURE.cancelled())
671	n/a	self.assertFalse(EXCEPTION_FUTURE.cancelled())
672	n/a	self.assertFalse(SUCCESSFUL_FUTURE.cancelled())
673	n/a
674	n/a	def test_done(self):
675	n/a	self.assertFalse(PENDING_FUTURE.done())
676	n/a	self.assertFalse(RUNNING_FUTURE.done())
677	n/a	self.assertTrue(CANCELLED_FUTURE.done())
678	n/a	self.assertTrue(CANCELLED_AND_NOTIFIED_FUTURE.done())
679	n/a	self.assertTrue(EXCEPTION_FUTURE.done())
680	n/a	self.assertTrue(SUCCESSFUL_FUTURE.done())
681	n/a
682	n/a	def test_running(self):
683	n/a	self.assertFalse(PENDING_FUTURE.running())
684	n/a	self.assertTrue(RUNNING_FUTURE.running())
685	n/a	self.assertFalse(CANCELLED_FUTURE.running())
686	n/a	self.assertFalse(CANCELLED_AND_NOTIFIED_FUTURE.running())
687	n/a	self.assertFalse(EXCEPTION_FUTURE.running())
688	n/a	self.assertFalse(SUCCESSFUL_FUTURE.running())
689	n/a
690	n/a	def test_result_with_timeout(self):
691	n/a	self.assertRaises(futures.TimeoutError,
692	n/a	PENDING_FUTURE.result, timeout=0)
693	n/a	self.assertRaises(futures.TimeoutError,
694	n/a	RUNNING_FUTURE.result, timeout=0)
695	n/a	self.assertRaises(futures.CancelledError,
696	n/a	CANCELLED_FUTURE.result, timeout=0)
697	n/a	self.assertRaises(futures.CancelledError,
698	n/a	CANCELLED_AND_NOTIFIED_FUTURE.result, timeout=0)
699	n/a	self.assertRaises(OSError, EXCEPTION_FUTURE.result, timeout=0)
700	n/a	self.assertEqual(SUCCESSFUL_FUTURE.result(timeout=0), 42)
701	n/a
702	n/a	def test_result_with_success(self):
703	n/a	# TODO(brian@sweetapp.com): This test is timing dependent.
704	n/a	def notification():
705	n/a	# Wait until the main thread is waiting for the result.
706	n/a	time.sleep(1)
707	n/a	f1.set_result(42)
708	n/a
709	n/a	f1 = create_future(state=PENDING)
710	n/a	t = threading.Thread(target=notification)
711	n/a	t.start()
712	n/a
713	n/a	self.assertEqual(f1.result(timeout=5), 42)
714	n/a
715	n/a	def test_result_with_cancel(self):
716	n/a	# TODO(brian@sweetapp.com): This test is timing dependent.
717	n/a	def notification():
718	n/a	# Wait until the main thread is waiting for the result.
719	n/a	time.sleep(1)
720	n/a	f1.cancel()
721	n/a
722	n/a	f1 = create_future(state=PENDING)
723	n/a	t = threading.Thread(target=notification)
724	n/a	t.start()
725	n/a
726	n/a	self.assertRaises(futures.CancelledError, f1.result, timeout=5)
727	n/a
728	n/a	def test_exception_with_timeout(self):
729	n/a	self.assertRaises(futures.TimeoutError,
730	n/a	PENDING_FUTURE.exception, timeout=0)
731	n/a	self.assertRaises(futures.TimeoutError,
732	n/a	RUNNING_FUTURE.exception, timeout=0)
733	n/a	self.assertRaises(futures.CancelledError,
734	n/a	CANCELLED_FUTURE.exception, timeout=0)
735	n/a	self.assertRaises(futures.CancelledError,
736	n/a	CANCELLED_AND_NOTIFIED_FUTURE.exception, timeout=0)
737	n/a	self.assertTrue(isinstance(EXCEPTION_FUTURE.exception(timeout=0),
738	n/a	OSError))
739	n/a	self.assertEqual(SUCCESSFUL_FUTURE.exception(timeout=0), None)
740	n/a
741	n/a	def test_exception_with_success(self):
742	n/a	def notification():
743	n/a	# Wait until the main thread is waiting for the exception.
744	n/a	time.sleep(1)
745	n/a	with f1._condition:
746	n/a	f1._state = FINISHED
747	n/a	f1._exception = OSError()
748	n/a	f1._condition.notify_all()
749	n/a
750	n/a	f1 = create_future(state=PENDING)
751	n/a	t = threading.Thread(target=notification)
752	n/a	t.start()
753	n/a
754	n/a	self.assertTrue(isinstance(f1.exception(timeout=5), OSError))
755	n/a
756	n/a	@test.support.reap_threads
757	n/a	def test_main():
758	n/a	try:
759	n/a	test.support.run_unittest(__name__)
760	n/a	finally:
761	n/a	test.support.reap_children()
762	n/a
763	n/a	if __name__ == "__main__":
764	n/a	test_main()