Python code coverage for Lib/test/test_contextlib.py

#	count	content
1	n/a	"""Unit tests for contextlib.py, and other context managers."""
2	n/a
3	n/a	import io
4	n/a	import sys
5	n/a	import tempfile
6	n/a	import unittest
7	n/a	from contextlib import * # Tests __all__
8	n/a	from test import support
9	n/a	try:
10	n/a	import threading
11	n/a	except ImportError:
12	n/a	threading = None
13	n/a
14	n/a
15	n/a	class TestAbstractContextManager(unittest.TestCase):
16	n/a
17	n/a	def test_enter(self):
18	n/a	class DefaultEnter(AbstractContextManager):
19	n/a	def __exit__(self, *args):
20	n/a	super().__exit__(*args)
21	n/a
22	n/a	manager = DefaultEnter()
23	n/a	self.assertIs(manager.__enter__(), manager)
24	n/a
25	n/a	def test_exit_is_abstract(self):
26	n/a	class MissingExit(AbstractContextManager):
27	n/a	pass
28	n/a
29	n/a	with self.assertRaises(TypeError):
30	n/a	MissingExit()
31	n/a
32	n/a	def test_structural_subclassing(self):
33	n/a	class ManagerFromScratch:
34	n/a	def __enter__(self):
35	n/a	return self
36	n/a	def __exit__(self, exc_type, exc_value, traceback):
37	n/a	return None
38	n/a
39	n/a	self.assertTrue(issubclass(ManagerFromScratch, AbstractContextManager))
40	n/a
41	n/a	class DefaultEnter(AbstractContextManager):
42	n/a	def __exit__(self, *args):
43	n/a	super().__exit__(*args)
44	n/a
45	n/a	self.assertTrue(issubclass(DefaultEnter, AbstractContextManager))
46	n/a
47	n/a
48	n/a	class ContextManagerTestCase(unittest.TestCase):
49	n/a
50	n/a	def test_contextmanager_plain(self):
51	n/a	state = []
52	n/a	@contextmanager
53	n/a	def woohoo():
54	n/a	state.append(1)
55	n/a	yield 42
56	n/a	state.append(999)
57	n/a	with woohoo() as x:
58	n/a	self.assertEqual(state, [1])
59	n/a	self.assertEqual(x, 42)
60	n/a	state.append(x)
61	n/a	self.assertEqual(state, [1, 42, 999])
62	n/a
63	n/a	def test_contextmanager_finally(self):
64	n/a	state = []
65	n/a	@contextmanager
66	n/a	def woohoo():
67	n/a	state.append(1)
68	n/a	try:
69	n/a	yield 42
70	n/a	finally:
71	n/a	state.append(999)
72	n/a	with self.assertRaises(ZeroDivisionError):
73	n/a	with woohoo() as x:
74	n/a	self.assertEqual(state, [1])
75	n/a	self.assertEqual(x, 42)
76	n/a	state.append(x)
77	n/a	raise ZeroDivisionError()
78	n/a	self.assertEqual(state, [1, 42, 999])
79	n/a
80	n/a	def test_contextmanager_no_reraise(self):
81	n/a	@contextmanager
82	n/a	def whee():
83	n/a	yield
84	n/a	ctx = whee()
85	n/a	ctx.__enter__()
86	n/a	# Calling __exit__ should not result in an exception
87	n/a	self.assertFalse(ctx.__exit__(TypeError, TypeError("foo"), None))
88	n/a
89	n/a	def test_contextmanager_trap_yield_after_throw(self):
90	n/a	@contextmanager
91	n/a	def whoo():
92	n/a	try:
93	n/a	yield
94	n/a	except:
95	n/a	yield
96	n/a	ctx = whoo()
97	n/a	ctx.__enter__()
98	n/a	self.assertRaises(
99	n/a	RuntimeError, ctx.__exit__, TypeError, TypeError("foo"), None
100	n/a	)
101	n/a
102	n/a	def test_contextmanager_except(self):
103	n/a	state = []
104	n/a	@contextmanager
105	n/a	def woohoo():
106	n/a	state.append(1)
107	n/a	try:
108	n/a	yield 42
109	n/a	except ZeroDivisionError as e:
110	n/a	state.append(e.args[0])
111	n/a	self.assertEqual(state, [1, 42, 999])
112	n/a	with woohoo() as x:
113	n/a	self.assertEqual(state, [1])
114	n/a	self.assertEqual(x, 42)
115	n/a	state.append(x)
116	n/a	raise ZeroDivisionError(999)
117	n/a	self.assertEqual(state, [1, 42, 999])
118	n/a
119	n/a	def test_contextmanager_except_stopiter(self):
120	n/a	stop_exc = StopIteration('spam')
121	n/a	@contextmanager
122	n/a	def woohoo():
123	n/a	yield
124	n/a	try:
125	n/a	with self.assertWarnsRegex(DeprecationWarning,
126	n/a	"StopIteration"):
127	n/a	with woohoo():
128	n/a	raise stop_exc
129	n/a	except Exception as ex:
130	n/a	self.assertIs(ex, stop_exc)
131	n/a	else:
132	n/a	self.fail('StopIteration was suppressed')
133	n/a
134	n/a	def test_contextmanager_except_pep479(self):
135	n/a	code = """\
136	n/a	from __future__ import generator_stop
137	n/a	from contextlib import contextmanager
138	n/a	@contextmanager
139	n/a	def woohoo():
140	n/a	yield
141	n/a	"""
142	n/a	locals = {}
143	n/a	exec(code, locals, locals)
144	n/a	woohoo = locals['woohoo']
145	n/a
146	n/a	stop_exc = StopIteration('spam')
147	n/a	try:
148	n/a	with woohoo():
149	n/a	raise stop_exc
150	n/a	except Exception as ex:
151	n/a	self.assertIs(ex, stop_exc)
152	n/a	else:
153	n/a	self.fail('StopIteration was suppressed')
154	n/a
155	n/a	def _create_contextmanager_attribs(self):
156	n/a	def attribs(**kw):
157	n/a	def decorate(func):
158	n/a	for k,v in kw.items():
159	n/a	setattr(func,k,v)
160	n/a	return func
161	n/a	return decorate
162	n/a	@contextmanager
163	n/a	@attribs(foo='bar')
164	n/a	def baz(spam):
165	n/a	"""Whee!"""
166	n/a	return baz
167	n/a
168	n/a	def test_contextmanager_attribs(self):
169	n/a	baz = self._create_contextmanager_attribs()
170	n/a	self.assertEqual(baz.__name__,'baz')
171	n/a	self.assertEqual(baz.foo, 'bar')
172	n/a
173	n/a	@support.requires_docstrings
174	n/a	def test_contextmanager_doc_attrib(self):
175	n/a	baz = self._create_contextmanager_attribs()
176	n/a	self.assertEqual(baz.__doc__, "Whee!")
177	n/a
178	n/a	@support.requires_docstrings
179	n/a	def test_instance_docstring_given_cm_docstring(self):
180	n/a	baz = self._create_contextmanager_attribs()(None)
181	n/a	self.assertEqual(baz.__doc__, "Whee!")
182	n/a
183	n/a	def test_keywords(self):
184	n/a	# Ensure no keyword arguments are inhibited
185	n/a	@contextmanager
186	n/a	def woohoo(self, func, args, kwds):
187	n/a	yield (self, func, args, kwds)
188	n/a	with woohoo(self=11, func=22, args=33, kwds=44) as target:
189	n/a	self.assertEqual(target, (11, 22, 33, 44))
190	n/a
191	n/a
192	n/a	class ClosingTestCase(unittest.TestCase):
193	n/a
194	n/a	@support.requires_docstrings
195	n/a	def test_instance_docs(self):
196	n/a	# Issue 19330: ensure context manager instances have good docstrings
197	n/a	cm_docstring = closing.__doc__
198	n/a	obj = closing(None)
199	n/a	self.assertEqual(obj.__doc__, cm_docstring)
200	n/a
201	n/a	def test_closing(self):
202	n/a	state = []
203	n/a	class C:
204	n/a	def close(self):
205	n/a	state.append(1)
206	n/a	x = C()
207	n/a	self.assertEqual(state, [])
208	n/a	with closing(x) as y:
209	n/a	self.assertEqual(x, y)
210	n/a	self.assertEqual(state, [1])
211	n/a
212	n/a	def test_closing_error(self):
213	n/a	state = []
214	n/a	class C:
215	n/a	def close(self):
216	n/a	state.append(1)
217	n/a	x = C()
218	n/a	self.assertEqual(state, [])
219	n/a	with self.assertRaises(ZeroDivisionError):
220	n/a	with closing(x) as y:
221	n/a	self.assertEqual(x, y)
222	n/a	1 / 0
223	n/a	self.assertEqual(state, [1])
224	n/a
225	n/a	class FileContextTestCase(unittest.TestCase):
226	n/a
227	n/a	def testWithOpen(self):
228	n/a	tfn = tempfile.mktemp()
229	n/a	try:
230	n/a	f = None
231	n/a	with open(tfn, "w") as f:
232	n/a	self.assertFalse(f.closed)
233	n/a	f.write("Booh\n")
234	n/a	self.assertTrue(f.closed)
235	n/a	f = None
236	n/a	with self.assertRaises(ZeroDivisionError):
237	n/a	with open(tfn, "r") as f:
238	n/a	self.assertFalse(f.closed)
239	n/a	self.assertEqual(f.read(), "Booh\n")
240	n/a	1 / 0
241	n/a	self.assertTrue(f.closed)
242	n/a	finally:
243	n/a	support.unlink(tfn)
244	n/a
245	n/a	@unittest.skipUnless(threading, 'Threading required for this test.')
246	n/a	class LockContextTestCase(unittest.TestCase):
247	n/a
248	n/a	def boilerPlate(self, lock, locked):
249	n/a	self.assertFalse(locked())
250	n/a	with lock:
251	n/a	self.assertTrue(locked())
252	n/a	self.assertFalse(locked())
253	n/a	with self.assertRaises(ZeroDivisionError):
254	n/a	with lock:
255	n/a	self.assertTrue(locked())
256	n/a	1 / 0
257	n/a	self.assertFalse(locked())
258	n/a
259	n/a	def testWithLock(self):
260	n/a	lock = threading.Lock()
261	n/a	self.boilerPlate(lock, lock.locked)
262	n/a
263	n/a	def testWithRLock(self):
264	n/a	lock = threading.RLock()
265	n/a	self.boilerPlate(lock, lock._is_owned)
266	n/a
267	n/a	def testWithCondition(self):
268	n/a	lock = threading.Condition()
269	n/a	def locked():
270	n/a	return lock._is_owned()
271	n/a	self.boilerPlate(lock, locked)
272	n/a
273	n/a	def testWithSemaphore(self):
274	n/a	lock = threading.Semaphore()
275	n/a	def locked():
276	n/a	if lock.acquire(False):
277	n/a	lock.release()
278	n/a	return False
279	n/a	else:
280	n/a	return True
281	n/a	self.boilerPlate(lock, locked)
282	n/a
283	n/a	def testWithBoundedSemaphore(self):
284	n/a	lock = threading.BoundedSemaphore()
285	n/a	def locked():
286	n/a	if lock.acquire(False):
287	n/a	lock.release()
288	n/a	return False
289	n/a	else:
290	n/a	return True
291	n/a	self.boilerPlate(lock, locked)
292	n/a
293	n/a
294	n/a	class mycontext(ContextDecorator):
295	n/a	"""Example decoration-compatible context manager for testing"""
296	n/a	started = False
297	n/a	exc = None
298	n/a	catch = False
299	n/a
300	n/a	def __enter__(self):
301	n/a	self.started = True
302	n/a	return self
303	n/a
304	n/a	def __exit__(self, *exc):
305	n/a	self.exc = exc
306	n/a	return self.catch
307	n/a
308	n/a
309	n/a	class TestContextDecorator(unittest.TestCase):
310	n/a
311	n/a	@support.requires_docstrings
312	n/a	def test_instance_docs(self):
313	n/a	# Issue 19330: ensure context manager instances have good docstrings
314	n/a	cm_docstring = mycontext.__doc__
315	n/a	obj = mycontext()
316	n/a	self.assertEqual(obj.__doc__, cm_docstring)
317	n/a
318	n/a	def test_contextdecorator(self):
319	n/a	context = mycontext()
320	n/a	with context as result:
321	n/a	self.assertIs(result, context)
322	n/a	self.assertTrue(context.started)
323	n/a
324	n/a	self.assertEqual(context.exc, (None, None, None))
325	n/a
326	n/a
327	n/a	def test_contextdecorator_with_exception(self):
328	n/a	context = mycontext()
329	n/a
330	n/a	with self.assertRaisesRegex(NameError, 'foo'):
331	n/a	with context:
332	n/a	raise NameError('foo')
333	n/a	self.assertIsNotNone(context.exc)
334	n/a	self.assertIs(context.exc[0], NameError)
335	n/a
336	n/a	context = mycontext()
337	n/a	context.catch = True
338	n/a	with context:
339	n/a	raise NameError('foo')
340	n/a	self.assertIsNotNone(context.exc)
341	n/a	self.assertIs(context.exc[0], NameError)
342	n/a
343	n/a
344	n/a	def test_decorator(self):
345	n/a	context = mycontext()
346	n/a
347	n/a	@context
348	n/a	def test():
349	n/a	self.assertIsNone(context.exc)
350	n/a	self.assertTrue(context.started)
351	n/a	test()
352	n/a	self.assertEqual(context.exc, (None, None, None))
353	n/a
354	n/a
355	n/a	def test_decorator_with_exception(self):
356	n/a	context = mycontext()
357	n/a
358	n/a	@context
359	n/a	def test():
360	n/a	self.assertIsNone(context.exc)
361	n/a	self.assertTrue(context.started)
362	n/a	raise NameError('foo')
363	n/a
364	n/a	with self.assertRaisesRegex(NameError, 'foo'):
365	n/a	test()
366	n/a	self.assertIsNotNone(context.exc)
367	n/a	self.assertIs(context.exc[0], NameError)
368	n/a
369	n/a
370	n/a	def test_decorating_method(self):
371	n/a	context = mycontext()
372	n/a
373	n/a	class Test(object):
374	n/a
375	n/a	@context
376	n/a	def method(self, a, b, c=None):
377	n/a	self.a = a
378	n/a	self.b = b
379	n/a	self.c = c
380	n/a
381	n/a	# these tests are for argument passing when used as a decorator
382	n/a	test = Test()
383	n/a	test.method(1, 2)
384	n/a	self.assertEqual(test.a, 1)
385	n/a	self.assertEqual(test.b, 2)
386	n/a	self.assertEqual(test.c, None)
387	n/a
388	n/a	test = Test()
389	n/a	test.method('a', 'b', 'c')
390	n/a	self.assertEqual(test.a, 'a')
391	n/a	self.assertEqual(test.b, 'b')
392	n/a	self.assertEqual(test.c, 'c')
393	n/a
394	n/a	test = Test()
395	n/a	test.method(a=1, b=2)
396	n/a	self.assertEqual(test.a, 1)
397	n/a	self.assertEqual(test.b, 2)
398	n/a
399	n/a
400	n/a	def test_typo_enter(self):
401	n/a	class mycontext(ContextDecorator):
402	n/a	def __unter__(self):
403	n/a	pass
404	n/a	def __exit__(self, *exc):
405	n/a	pass
406	n/a
407	n/a	with self.assertRaises(AttributeError):
408	n/a	with mycontext():
409	n/a	pass
410	n/a
411	n/a
412	n/a	def test_typo_exit(self):
413	n/a	class mycontext(ContextDecorator):
414	n/a	def __enter__(self):
415	n/a	pass
416	n/a	def __uxit__(self, *exc):
417	n/a	pass
418	n/a
419	n/a	with self.assertRaises(AttributeError):
420	n/a	with mycontext():
421	n/a	pass
422	n/a
423	n/a
424	n/a	def test_contextdecorator_as_mixin(self):
425	n/a	class somecontext(object):
426	n/a	started = False
427	n/a	exc = None
428	n/a
429	n/a	def __enter__(self):
430	n/a	self.started = True
431	n/a	return self
432	n/a
433	n/a	def __exit__(self, *exc):
434	n/a	self.exc = exc
435	n/a
436	n/a	class mycontext(somecontext, ContextDecorator):
437	n/a	pass
438	n/a
439	n/a	context = mycontext()
440	n/a	@context
441	n/a	def test():
442	n/a	self.assertIsNone(context.exc)
443	n/a	self.assertTrue(context.started)
444	n/a	test()
445	n/a	self.assertEqual(context.exc, (None, None, None))
446	n/a
447	n/a
448	n/a	def test_contextmanager_as_decorator(self):
449	n/a	@contextmanager
450	n/a	def woohoo(y):
451	n/a	state.append(y)
452	n/a	yield
453	n/a	state.append(999)
454	n/a
455	n/a	state = []
456	n/a	@woohoo(1)
457	n/a	def test(x):
458	n/a	self.assertEqual(state, [1])
459	n/a	state.append(x)
460	n/a	test('something')
461	n/a	self.assertEqual(state, [1, 'something', 999])
462	n/a
463	n/a	# Issue #11647: Ensure the decorated function is 'reusable'
464	n/a	state = []
465	n/a	test('something else')
466	n/a	self.assertEqual(state, [1, 'something else', 999])
467	n/a
468	n/a
469	n/a	class TestExitStack(unittest.TestCase):
470	n/a
471	n/a	@support.requires_docstrings
472	n/a	def test_instance_docs(self):
473	n/a	# Issue 19330: ensure context manager instances have good docstrings
474	n/a	cm_docstring = ExitStack.__doc__
475	n/a	obj = ExitStack()
476	n/a	self.assertEqual(obj.__doc__, cm_docstring)
477	n/a
478	n/a	def test_no_resources(self):
479	n/a	with ExitStack():
480	n/a	pass
481	n/a
482	n/a	def test_callback(self):
483	n/a	expected = [
484	n/a	((), {}),
485	n/a	((1,), {}),
486	n/a	((1,2), {}),
487	n/a	((), dict(example=1)),
488	n/a	((1,), dict(example=1)),
489	n/a	((1,2), dict(example=1)),
490	n/a	]
491	n/a	result = []
492	n/a	def _exit(args, *kwds):
493	n/a	"""Test metadata propagation"""
494	n/a	result.append((args, kwds))
495	n/a	with ExitStack() as stack:
496	n/a	for args, kwds in reversed(expected):
497	n/a	if args and kwds:
498	n/a	f = stack.callback(_exit, args, *kwds)
499	n/a	elif args:
500	n/a	f = stack.callback(_exit, *args)
501	n/a	elif kwds:
502	n/a	f = stack.callback(_exit, **kwds)
503	n/a	else:
504	n/a	f = stack.callback(_exit)
505	n/a	self.assertIs(f, _exit)
506	n/a	for wrapper in stack._exit_callbacks:
507	n/a	self.assertIs(wrapper.__wrapped__, _exit)
508	n/a	self.assertNotEqual(wrapper.__name__, _exit.__name__)
509	n/a	self.assertIsNone(wrapper.__doc__, _exit.__doc__)
510	n/a	self.assertEqual(result, expected)
511	n/a
512	n/a	def test_push(self):
513	n/a	exc_raised = ZeroDivisionError
514	n/a	def _expect_exc(exc_type, exc, exc_tb):
515	n/a	self.assertIs(exc_type, exc_raised)
516	n/a	def _suppress_exc(*exc_details):
517	n/a	return True
518	n/a	def _expect_ok(exc_type, exc, exc_tb):
519	n/a	self.assertIsNone(exc_type)
520	n/a	self.assertIsNone(exc)
521	n/a	self.assertIsNone(exc_tb)
522	n/a	class ExitCM(object):
523	n/a	def __init__(self, check_exc):
524	n/a	self.check_exc = check_exc
525	n/a	def __enter__(self):
526	n/a	self.fail("Should not be called!")
527	n/a	def __exit__(self, *exc_details):
528	n/a	self.check_exc(*exc_details)
529	n/a	with ExitStack() as stack:
530	n/a	stack.push(_expect_ok)
531	n/a	self.assertIs(stack._exit_callbacks[-1], _expect_ok)
532	n/a	cm = ExitCM(_expect_ok)
533	n/a	stack.push(cm)
534	n/a	self.assertIs(stack._exit_callbacks[-1].__self__, cm)
535	n/a	stack.push(_suppress_exc)
536	n/a	self.assertIs(stack._exit_callbacks[-1], _suppress_exc)
537	n/a	cm = ExitCM(_expect_exc)
538	n/a	stack.push(cm)
539	n/a	self.assertIs(stack._exit_callbacks[-1].__self__, cm)
540	n/a	stack.push(_expect_exc)
541	n/a	self.assertIs(stack._exit_callbacks[-1], _expect_exc)
542	n/a	stack.push(_expect_exc)
543	n/a	self.assertIs(stack._exit_callbacks[-1], _expect_exc)
544	n/a	1/0
545	n/a
546	n/a	def test_enter_context(self):
547	n/a	class TestCM(object):
548	n/a	def __enter__(self):
549	n/a	result.append(1)
550	n/a	def __exit__(self, *exc_details):
551	n/a	result.append(3)
552	n/a
553	n/a	result = []
554	n/a	cm = TestCM()
555	n/a	with ExitStack() as stack:
556	n/a	@stack.callback # Registered first => cleaned up last
557	n/a	def _exit():
558	n/a	result.append(4)
559	n/a	self.assertIsNotNone(_exit)
560	n/a	stack.enter_context(cm)
561	n/a	self.assertIs(stack._exit_callbacks[-1].__self__, cm)
562	n/a	result.append(2)
563	n/a	self.assertEqual(result, [1, 2, 3, 4])
564	n/a
565	n/a	def test_close(self):
566	n/a	result = []
567	n/a	with ExitStack() as stack:
568	n/a	@stack.callback
569	n/a	def _exit():
570	n/a	result.append(1)
571	n/a	self.assertIsNotNone(_exit)
572	n/a	stack.close()
573	n/a	result.append(2)
574	n/a	self.assertEqual(result, [1, 2])
575	n/a
576	n/a	def test_pop_all(self):
577	n/a	result = []
578	n/a	with ExitStack() as stack:
579	n/a	@stack.callback
580	n/a	def _exit():
581	n/a	result.append(3)
582	n/a	self.assertIsNotNone(_exit)
583	n/a	new_stack = stack.pop_all()
584	n/a	result.append(1)
585	n/a	result.append(2)
586	n/a	new_stack.close()
587	n/a	self.assertEqual(result, [1, 2, 3])
588	n/a
589	n/a	def test_exit_raise(self):
590	n/a	with self.assertRaises(ZeroDivisionError):
591	n/a	with ExitStack() as stack:
592	n/a	stack.push(lambda *exc: False)
593	n/a	1/0
594	n/a
595	n/a	def test_exit_suppress(self):
596	n/a	with ExitStack() as stack:
597	n/a	stack.push(lambda *exc: True)
598	n/a	1/0
599	n/a
600	n/a	def test_exit_exception_chaining_reference(self):
601	n/a	# Sanity check to make sure that ExitStack chaining matches
602	n/a	# actual nested with statements
603	n/a	class RaiseExc:
604	n/a	def __init__(self, exc):
605	n/a	self.exc = exc
606	n/a	def __enter__(self):
607	n/a	return self
608	n/a	def __exit__(self, *exc_details):
609	n/a	raise self.exc
610	n/a
611	n/a	class RaiseExcWithContext:
612	n/a	def __init__(self, outer, inner):
613	n/a	self.outer = outer
614	n/a	self.inner = inner
615	n/a	def __enter__(self):
616	n/a	return self
617	n/a	def __exit__(self, *exc_details):
618	n/a	try:
619	n/a	raise self.inner
620	n/a	except:
621	n/a	raise self.outer
622	n/a
623	n/a	class SuppressExc:
624	n/a	def __enter__(self):
625	n/a	return self
626	n/a	def __exit__(self, *exc_details):
627	n/a	type(self).saved_details = exc_details
628	n/a	return True
629	n/a
630	n/a	try:
631	n/a	with RaiseExc(IndexError):
632	n/a	with RaiseExcWithContext(KeyError, AttributeError):
633	n/a	with SuppressExc():
634	n/a	with RaiseExc(ValueError):
635	n/a	1 / 0
636	n/a	except IndexError as exc:
637	n/a	self.assertIsInstance(exc.__context__, KeyError)
638	n/a	self.assertIsInstance(exc.__context__.__context__, AttributeError)
639	n/a	# Inner exceptions were suppressed
640	n/a	self.assertIsNone(exc.__context__.__context__.__context__)
641	n/a	else:
642	n/a	self.fail("Expected IndexError, but no exception was raised")
643	n/a	# Check the inner exceptions
644	n/a	inner_exc = SuppressExc.saved_details[1]
645	n/a	self.assertIsInstance(inner_exc, ValueError)
646	n/a	self.assertIsInstance(inner_exc.__context__, ZeroDivisionError)
647	n/a
648	n/a	def test_exit_exception_chaining(self):
649	n/a	# Ensure exception chaining matches the reference behaviour
650	n/a	def raise_exc(exc):
651	n/a	raise exc
652	n/a
653	n/a	saved_details = None
654	n/a	def suppress_exc(*exc_details):
655	n/a	nonlocal saved_details
656	n/a	saved_details = exc_details
657	n/a	return True
658	n/a
659	n/a	try:
660	n/a	with ExitStack() as stack:
661	n/a	stack.callback(raise_exc, IndexError)
662	n/a	stack.callback(raise_exc, KeyError)
663	n/a	stack.callback(raise_exc, AttributeError)
664	n/a	stack.push(suppress_exc)
665	n/a	stack.callback(raise_exc, ValueError)
666	n/a	1 / 0
667	n/a	except IndexError as exc:
668	n/a	self.assertIsInstance(exc.__context__, KeyError)
669	n/a	self.assertIsInstance(exc.__context__.__context__, AttributeError)
670	n/a	# Inner exceptions were suppressed
671	n/a	self.assertIsNone(exc.__context__.__context__.__context__)
672	n/a	else:
673	n/a	self.fail("Expected IndexError, but no exception was raised")
674	n/a	# Check the inner exceptions
675	n/a	inner_exc = saved_details[1]
676	n/a	self.assertIsInstance(inner_exc, ValueError)
677	n/a	self.assertIsInstance(inner_exc.__context__, ZeroDivisionError)
678	n/a
679	n/a	def test_exit_exception_non_suppressing(self):
680	n/a	# http://bugs.python.org/issue19092
681	n/a	def raise_exc(exc):
682	n/a	raise exc
683	n/a
684	n/a	def suppress_exc(*exc_details):
685	n/a	return True
686	n/a
687	n/a	try:
688	n/a	with ExitStack() as stack:
689	n/a	stack.callback(lambda: None)
690	n/a	stack.callback(raise_exc, IndexError)
691	n/a	except Exception as exc:
692	n/a	self.assertIsInstance(exc, IndexError)
693	n/a	else:
694	n/a	self.fail("Expected IndexError, but no exception was raised")
695	n/a
696	n/a	try:
697	n/a	with ExitStack() as stack:
698	n/a	stack.callback(raise_exc, KeyError)
699	n/a	stack.push(suppress_exc)
700	n/a	stack.callback(raise_exc, IndexError)
701	n/a	except Exception as exc:
702	n/a	self.assertIsInstance(exc, KeyError)
703	n/a	else:
704	n/a	self.fail("Expected KeyError, but no exception was raised")
705	n/a
706	n/a	def test_exit_exception_with_correct_context(self):
707	n/a	# http://bugs.python.org/issue20317
708	n/a	@contextmanager
709	n/a	def gets_the_context_right(exc):
710	n/a	try:
711	n/a	yield
712	n/a	finally:
713	n/a	raise exc
714	n/a
715	n/a	exc1 = Exception(1)
716	n/a	exc2 = Exception(2)
717	n/a	exc3 = Exception(3)
718	n/a	exc4 = Exception(4)
719	n/a
720	n/a	# The contextmanager already fixes the context, so prior to the
721	n/a	# fix, ExitStack would try to fix it again and get into an
722	n/a	# infinite self-referential loop
723	n/a	try:
724	n/a	with ExitStack() as stack:
725	n/a	stack.enter_context(gets_the_context_right(exc4))
726	n/a	stack.enter_context(gets_the_context_right(exc3))
727	n/a	stack.enter_context(gets_the_context_right(exc2))
728	n/a	raise exc1
729	n/a	except Exception as exc:
730	n/a	self.assertIs(exc, exc4)
731	n/a	self.assertIs(exc.__context__, exc3)
732	n/a	self.assertIs(exc.__context__.__context__, exc2)
733	n/a	self.assertIs(exc.__context__.__context__.__context__, exc1)
734	n/a	self.assertIsNone(
735	n/a	exc.__context__.__context__.__context__.__context__)
736	n/a
737	n/a	def test_exit_exception_with_existing_context(self):
738	n/a	# Addresses a lack of test coverage discovered after checking in a
739	n/a	# fix for issue 20317 that still contained debugging code.
740	n/a	def raise_nested(inner_exc, outer_exc):
741	n/a	try:
742	n/a	raise inner_exc
743	n/a	finally:
744	n/a	raise outer_exc
745	n/a	exc1 = Exception(1)
746	n/a	exc2 = Exception(2)
747	n/a	exc3 = Exception(3)
748	n/a	exc4 = Exception(4)
749	n/a	exc5 = Exception(5)
750	n/a	try:
751	n/a	with ExitStack() as stack:
752	n/a	stack.callback(raise_nested, exc4, exc5)
753	n/a	stack.callback(raise_nested, exc2, exc3)
754	n/a	raise exc1
755	n/a	except Exception as exc:
756	n/a	self.assertIs(exc, exc5)
757	n/a	self.assertIs(exc.__context__, exc4)
758	n/a	self.assertIs(exc.__context__.__context__, exc3)
759	n/a	self.assertIs(exc.__context__.__context__.__context__, exc2)
760	n/a	self.assertIs(
761	n/a	exc.__context__.__context__.__context__.__context__, exc1)
762	n/a	self.assertIsNone(
763	n/a	exc.__context__.__context__.__context__.__context__.__context__)
764	n/a
765	n/a
766	n/a
767	n/a	def test_body_exception_suppress(self):
768	n/a	def suppress_exc(*exc_details):
769	n/a	return True
770	n/a	try:
771	n/a	with ExitStack() as stack:
772	n/a	stack.push(suppress_exc)
773	n/a	1/0
774	n/a	except IndexError as exc:
775	n/a	self.fail("Expected no exception, got IndexError")
776	n/a
777	n/a	def test_exit_exception_chaining_suppress(self):
778	n/a	with ExitStack() as stack:
779	n/a	stack.push(lambda *exc: True)
780	n/a	stack.push(lambda *exc: 1/0)
781	n/a	stack.push(lambda *exc: {}[1])
782	n/a
783	n/a	def test_excessive_nesting(self):
784	n/a	# The original implementation would die with RecursionError here
785	n/a	with ExitStack() as stack:
786	n/a	for i in range(10000):
787	n/a	stack.callback(int)
788	n/a
789	n/a	def test_instance_bypass(self):
790	n/a	class Example(object): pass
791	n/a	cm = Example()
792	n/a	cm.__exit__ = object()
793	n/a	stack = ExitStack()
794	n/a	self.assertRaises(AttributeError, stack.enter_context, cm)
795	n/a	stack.push(cm)
796	n/a	self.assertIs(stack._exit_callbacks[-1], cm)
797	n/a
798	n/a	def test_dont_reraise_RuntimeError(self):
799	n/a	# https://bugs.python.org/issue27122
800	n/a	class UniqueException(Exception): pass
801	n/a	class UniqueRuntimeError(RuntimeError): pass
802	n/a
803	n/a	@contextmanager
804	n/a	def second():
805	n/a	try:
806	n/a	yield 1
807	n/a	except Exception as exc:
808	n/a	raise UniqueException("new exception") from exc
809	n/a
810	n/a	@contextmanager
811	n/a	def first():
812	n/a	try:
813	n/a	yield 1
814	n/a	except Exception as exc:
815	n/a	raise exc
816	n/a
817	n/a	# The UniqueRuntimeError should be caught by second()'s exception
818	n/a	# handler which chain raised a new UniqueException.
819	n/a	with self.assertRaises(UniqueException) as err_ctx:
820	n/a	with ExitStack() as es_ctx:
821	n/a	es_ctx.enter_context(second())
822	n/a	es_ctx.enter_context(first())
823	n/a	raise UniqueRuntimeError("please no infinite loop.")
824	n/a
825	n/a	exc = err_ctx.exception
826	n/a	self.assertIsInstance(exc, UniqueException)
827	n/a	self.assertIsInstance(exc.__context__, UniqueRuntimeError)
828	n/a	self.assertIsNone(exc.__context__.__context__)
829	n/a	self.assertIsNone(exc.__context__.__cause__)
830	n/a	self.assertIs(exc.__cause__, exc.__context__)
831	n/a
832	n/a
833	n/a	class TestRedirectStream:
834	n/a
835	n/a	redirect_stream = None
836	n/a	orig_stream = None
837	n/a
838	n/a	@support.requires_docstrings
839	n/a	def test_instance_docs(self):
840	n/a	# Issue 19330: ensure context manager instances have good docstrings
841	n/a	cm_docstring = self.redirect_stream.__doc__
842	n/a	obj = self.redirect_stream(None)
843	n/a	self.assertEqual(obj.__doc__, cm_docstring)
844	n/a
845	n/a	def test_no_redirect_in_init(self):
846	n/a	orig_stdout = getattr(sys, self.orig_stream)
847	n/a	self.redirect_stream(None)
848	n/a	self.assertIs(getattr(sys, self.orig_stream), orig_stdout)
849	n/a
850	n/a	def test_redirect_to_string_io(self):
851	n/a	f = io.StringIO()
852	n/a	msg = "Consider an API like help(), which prints directly to stdout"
853	n/a	orig_stdout = getattr(sys, self.orig_stream)
854	n/a	with self.redirect_stream(f):
855	n/a	print(msg, file=getattr(sys, self.orig_stream))
856	n/a	self.assertIs(getattr(sys, self.orig_stream), orig_stdout)
857	n/a	s = f.getvalue().strip()
858	n/a	self.assertEqual(s, msg)
859	n/a
860	n/a	def test_enter_result_is_target(self):
861	n/a	f = io.StringIO()
862	n/a	with self.redirect_stream(f) as enter_result:
863	n/a	self.assertIs(enter_result, f)
864	n/a
865	n/a	def test_cm_is_reusable(self):
866	n/a	f = io.StringIO()
867	n/a	write_to_f = self.redirect_stream(f)
868	n/a	orig_stdout = getattr(sys, self.orig_stream)
869	n/a	with write_to_f:
870	n/a	print("Hello", end=" ", file=getattr(sys, self.orig_stream))
871	n/a	with write_to_f:
872	n/a	print("World!", file=getattr(sys, self.orig_stream))
873	n/a	self.assertIs(getattr(sys, self.orig_stream), orig_stdout)
874	n/a	s = f.getvalue()
875	n/a	self.assertEqual(s, "Hello World!\n")
876	n/a
877	n/a	def test_cm_is_reentrant(self):
878	n/a	f = io.StringIO()
879	n/a	write_to_f = self.redirect_stream(f)
880	n/a	orig_stdout = getattr(sys, self.orig_stream)
881	n/a	with write_to_f:
882	n/a	print("Hello", end=" ", file=getattr(sys, self.orig_stream))
883	n/a	with write_to_f:
884	n/a	print("World!", file=getattr(sys, self.orig_stream))
885	n/a	self.assertIs(getattr(sys, self.orig_stream), orig_stdout)
886	n/a	s = f.getvalue()
887	n/a	self.assertEqual(s, "Hello World!\n")
888	n/a
889	n/a
890	n/a	class TestRedirectStdout(TestRedirectStream, unittest.TestCase):
891	n/a
892	n/a	redirect_stream = redirect_stdout
893	n/a	orig_stream = "stdout"
894	n/a
895	n/a
896	n/a	class TestRedirectStderr(TestRedirectStream, unittest.TestCase):
897	n/a
898	n/a	redirect_stream = redirect_stderr
899	n/a	orig_stream = "stderr"
900	n/a
901	n/a
902	n/a	class TestSuppress(unittest.TestCase):
903	n/a
904	n/a	@support.requires_docstrings
905	n/a	def test_instance_docs(self):
906	n/a	# Issue 19330: ensure context manager instances have good docstrings
907	n/a	cm_docstring = suppress.__doc__
908	n/a	obj = suppress()
909	n/a	self.assertEqual(obj.__doc__, cm_docstring)
910	n/a
911	n/a	def test_no_result_from_enter(self):
912	n/a	with suppress(ValueError) as enter_result:
913	n/a	self.assertIsNone(enter_result)
914	n/a
915	n/a	def test_no_exception(self):
916	n/a	with suppress(ValueError):
917	n/a	self.assertEqual(pow(2, 5), 32)
918	n/a
919	n/a	def test_exact_exception(self):
920	n/a	with suppress(TypeError):
921	n/a	len(5)
922	n/a
923	n/a	def test_exception_hierarchy(self):
924	n/a	with suppress(LookupError):
925	n/a	'Hello'[50]
926	n/a
927	n/a	def test_other_exception(self):
928	n/a	with self.assertRaises(ZeroDivisionError):
929	n/a	with suppress(TypeError):
930	n/a	1/0
931	n/a
932	n/a	def test_no_args(self):
933	n/a	with self.assertRaises(ZeroDivisionError):
934	n/a	with suppress():
935	n/a	1/0
936	n/a
937	n/a	def test_multiple_exception_args(self):
938	n/a	with suppress(ZeroDivisionError, TypeError):
939	n/a	1/0
940	n/a	with suppress(ZeroDivisionError, TypeError):
941	n/a	len(5)
942	n/a
943	n/a	def test_cm_is_reentrant(self):
944	n/a	ignore_exceptions = suppress(Exception)
945	n/a	with ignore_exceptions:
946	n/a	pass
947	n/a	with ignore_exceptions:
948	n/a	len(5)
949	n/a	with ignore_exceptions:
950	n/a	with ignore_exceptions: # Check nested usage
951	n/a	len(5)
952	n/a	outer_continued = True
953	n/a	1/0
954	n/a	self.assertTrue(outer_continued)
955	n/a
956	n/a	if __name__ == "__main__":
957	n/a	unittest.main()