"""Unit tests for contextlib.py, and other context managers.""" import io import sys import tempfile import unittest from contextlib import * # Tests __all__ from test import support try: import threading except ImportError: threading = None class TestAbstractContextManager(unittest.TestCase): def test_enter(self): class DefaultEnter(AbstractContextManager): def __exit__(self, *args): super().__exit__(*args) manager = DefaultEnter() self.assertIs(manager.__enter__(), manager) def test_exit_is_abstract(self): class MissingExit(AbstractContextManager): pass with self.assertRaises(TypeError): MissingExit() def test_structural_subclassing(self): class ManagerFromScratch: def __enter__(self): return self def __exit__(self, exc_type, exc_value, traceback): return None self.assertTrue(issubclass(ManagerFromScratch, AbstractContextManager)) class DefaultEnter(AbstractContextManager): def __exit__(self, *args): super().__exit__(*args) self.assertTrue(issubclass(DefaultEnter, AbstractContextManager)) class ContextManagerTestCase(unittest.TestCase): def test_contextmanager_plain(self): state = [] @contextmanager def woohoo(): state.append(1) yield 42 state.append(999) with woohoo() as x: self.assertEqual(state, [1]) self.assertEqual(x, 42) state.append(x) self.assertEqual(state, [1, 42, 999]) def test_contextmanager_finally(self): state = [] @contextmanager def woohoo(): state.append(1) try: yield 42 finally: state.append(999) with self.assertRaises(ZeroDivisionError): with woohoo() as x: self.assertEqual(state, [1]) self.assertEqual(x, 42) state.append(x) raise ZeroDivisionError() self.assertEqual(state, [1, 42, 999]) def test_contextmanager_no_reraise(self): @contextmanager def whee(): yield ctx = whee() ctx.__enter__() # Calling __exit__ should not result in an exception self.assertFalse(ctx.__exit__(TypeError, TypeError("foo"), None)) def test_contextmanager_trap_yield_after_throw(self): @contextmanager def whoo(): try: yield except: yield ctx = whoo() ctx.__enter__() self.assertRaises( RuntimeError, ctx.__exit__, TypeError, TypeError("foo"), None ) def test_contextmanager_except(self): state = [] @contextmanager def woohoo(): state.append(1) try: yield 42 except ZeroDivisionError as e: state.append(e.args[0]) self.assertEqual(state, [1, 42, 999]) with woohoo() as x: self.assertEqual(state, [1]) self.assertEqual(x, 42) state.append(x) raise ZeroDivisionError(999) self.assertEqual(state, [1, 42, 999]) def test_contextmanager_except_stopiter(self): stop_exc = StopIteration('spam') @contextmanager def woohoo(): yield try: with self.assertWarnsRegex(DeprecationWarning, "StopIteration"): with woohoo(): raise stop_exc except Exception as ex: self.assertIs(ex, stop_exc) else: self.fail('StopIteration was suppressed') def test_contextmanager_except_pep479(self): code = """\ from __future__ import generator_stop from contextlib import contextmanager @contextmanager def woohoo(): yield """ locals = {} exec(code, locals, locals) woohoo = locals['woohoo'] stop_exc = StopIteration('spam') try: with woohoo(): raise stop_exc except Exception as ex: self.assertIs(ex, stop_exc) else: self.fail('StopIteration was suppressed') def test_contextmanager_do_not_unchain_non_stopiteration_exceptions(self): @contextmanager def test_issue29692(): try: yield except Exception as exc: raise RuntimeError('issue29692:Chained') from exc try: with test_issue29692(): raise ZeroDivisionError except Exception as ex: self.assertIs(type(ex), RuntimeError) self.assertEqual(ex.args[0], 'issue29692:Chained') self.assertIsInstance(ex.__cause__, ZeroDivisionError) try: with test_issue29692(): raise StopIteration('issue29692:Unchained') except Exception as ex: self.assertIs(type(ex), StopIteration) self.assertEqual(ex.args[0], 'issue29692:Unchained') self.assertIsNone(ex.__cause__) def _create_contextmanager_attribs(self): def attribs(**kw): def decorate(func): for k,v in kw.items(): setattr(func,k,v) return func return decorate @contextmanager @attribs(foo='bar') def baz(spam): """Whee!""" return baz def test_contextmanager_attribs(self): baz = self._create_contextmanager_attribs() self.assertEqual(baz.__name__,'baz') self.assertEqual(baz.foo, 'bar') @support.requires_docstrings def test_contextmanager_doc_attrib(self): baz = self._create_contextmanager_attribs() self.assertEqual(baz.__doc__, "Whee!") @support.requires_docstrings def test_instance_docstring_given_cm_docstring(self): baz = self._create_contextmanager_attribs()(None) self.assertEqual(baz.__doc__, "Whee!") def test_keywords(self): # Ensure no keyword arguments are inhibited @contextmanager def woohoo(self, func, args, kwds): yield (self, func, args, kwds) with woohoo(self=11, func=22, args=33, kwds=44) as target: self.assertEqual(target, (11, 22, 33, 44)) class ClosingTestCase(unittest.TestCase): @support.requires_docstrings def test_instance_docs(self): # Issue 19330: ensure context manager instances have good docstrings cm_docstring = closing.__doc__ obj = closing(None) self.assertEqual(obj.__doc__, cm_docstring) def test_closing(self): state = [] class C: def close(self): state.append(1) x = C() self.assertEqual(state, []) with closing(x) as y: self.assertEqual(x, y) self.assertEqual(state, [1]) def test_closing_error(self): state = [] class C: def close(self): state.append(1) x = C() self.assertEqual(state, []) with self.assertRaises(ZeroDivisionError): with closing(x) as y: self.assertEqual(x, y) 1 / 0 self.assertEqual(state, [1]) class FileContextTestCase(unittest.TestCase): def testWithOpen(self): tfn = tempfile.mktemp() try: f = None with open(tfn, "w") as f: self.assertFalse(f.closed) f.write("Booh\n") self.assertTrue(f.closed) f = None with self.assertRaises(ZeroDivisionError): with open(tfn, "r") as f: self.assertFalse(f.closed) self.assertEqual(f.read(), "Booh\n") 1 / 0 self.assertTrue(f.closed) finally: support.unlink(tfn) @unittest.skipUnless(threading, 'Threading required for this test.') class LockContextTestCase(unittest.TestCase): def boilerPlate(self, lock, locked): self.assertFalse(locked()) with lock: self.assertTrue(locked()) self.assertFalse(locked()) with self.assertRaises(ZeroDivisionError): with lock: self.assertTrue(locked()) 1 / 0 self.assertFalse(locked()) def testWithLock(self): lock = threading.Lock() self.boilerPlate(lock, lock.locked) def testWithRLock(self): lock = threading.RLock() self.boilerPlate(lock, lock._is_owned) def testWithCondition(self): lock = threading.Condition() def locked(): return lock._is_owned() self.boilerPlate(lock, locked) def testWithSemaphore(self): lock = threading.Semaphore() def locked(): if lock.acquire(False): lock.release() return False else: return True self.boilerPlate(lock, locked) def testWithBoundedSemaphore(self): lock = threading.BoundedSemaphore() def locked(): if lock.acquire(False): lock.release() return False else: return True self.boilerPlate(lock, locked) class mycontext(ContextDecorator): """Example decoration-compatible context manager for testing""" started = False exc = None catch = False def __enter__(self): self.started = True return self def __exit__(self, *exc): self.exc = exc return self.catch class TestContextDecorator(unittest.TestCase): @support.requires_docstrings def test_instance_docs(self): # Issue 19330: ensure context manager instances have good docstrings cm_docstring = mycontext.__doc__ obj = mycontext() self.assertEqual(obj.__doc__, cm_docstring) def test_contextdecorator(self): context = mycontext() with context as result: self.assertIs(result, context) self.assertTrue(context.started) self.assertEqual(context.exc, (None, None, None)) def test_contextdecorator_with_exception(self): context = mycontext() with self.assertRaisesRegex(NameError, 'foo'): with context: raise NameError('foo') self.assertIsNotNone(context.exc) self.assertIs(context.exc[0], NameError) context = mycontext() context.catch = True with context: raise NameError('foo') self.assertIsNotNone(context.exc) self.assertIs(context.exc[0], NameError) def test_decorator(self): context = mycontext() @context def test(): self.assertIsNone(context.exc) self.assertTrue(context.started) test() self.assertEqual(context.exc, (None, None, None)) def test_decorator_with_exception(self): context = mycontext() @context def test(): self.assertIsNone(context.exc) self.assertTrue(context.started) raise NameError('foo') with self.assertRaisesRegex(NameError, 'foo'): test() self.assertIsNotNone(context.exc) self.assertIs(context.exc[0], NameError) def test_decorating_method(self): context = mycontext() class Test(object): @context def method(self, a, b, c=None): self.a = a self.b = b self.c = c # these tests are for argument passing when used as a decorator test = Test() test.method(1, 2) self.assertEqual(test.a, 1) self.assertEqual(test.b, 2) self.assertEqual(test.c, None) test = Test() test.method('a', 'b', 'c') self.assertEqual(test.a, 'a') self.assertEqual(test.b, 'b') self.assertEqual(test.c, 'c') test = Test() test.method(a=1, b=2) self.assertEqual(test.a, 1) self.assertEqual(test.b, 2) def test_typo_enter(self): class mycontext(ContextDecorator): def __unter__(self): pass def __exit__(self, *exc): pass with self.assertRaises(AttributeError): with mycontext(): pass def test_typo_exit(self): class mycontext(ContextDecorator): def __enter__(self): pass def __uxit__(self, *exc): pass with self.assertRaises(AttributeError): with mycontext(): pass def test_contextdecorator_as_mixin(self): class somecontext(object): started = False exc = None def __enter__(self): self.started = True return self def __exit__(self, *exc): self.exc = exc class mycontext(somecontext, ContextDecorator): pass context = mycontext() @context def test(): self.assertIsNone(context.exc) self.assertTrue(context.started) test() self.assertEqual(context.exc, (None, None, None)) def test_contextmanager_as_decorator(self): @contextmanager def woohoo(y): state.append(y) yield state.append(999) state = [] @woohoo(1) def test(x): self.assertEqual(state, [1]) state.append(x) test('something') self.assertEqual(state, [1, 'something', 999]) # Issue #11647: Ensure the decorated function is 'reusable' state = [] test('something else') self.assertEqual(state, [1, 'something else', 999]) class TestExitStack(unittest.TestCase): @support.requires_docstrings def test_instance_docs(self): # Issue 19330: ensure context manager instances have good docstrings cm_docstring = ExitStack.__doc__ obj = ExitStack() self.assertEqual(obj.__doc__, cm_docstring) def test_no_resources(self): with ExitStack(): pass def test_callback(self): expected = [ ((), {}), ((1,), {}), ((1,2), {}), ((), dict(example=1)), ((1,), dict(example=1)), ((1,2), dict(example=1)), ] result = [] def _exit(*args, **kwds): """Test metadata propagation""" result.append((args, kwds)) with ExitStack() as stack: for args, kwds in reversed(expected): if args and kwds: f = stack.callback(_exit, *args, **kwds) elif args: f = stack.callback(_exit, *args) elif kwds: f = stack.callback(_exit, **kwds) else: f = stack.callback(_exit) self.assertIs(f, _exit) for wrapper in stack._exit_callbacks: self.assertIs(wrapper.__wrapped__, _exit) self.assertNotEqual(wrapper.__name__, _exit.__name__) self.assertIsNone(wrapper.__doc__, _exit.__doc__) self.assertEqual(result, expected) def test_push(self): exc_raised = ZeroDivisionError def _expect_exc(exc_type, exc, exc_tb): self.assertIs(exc_type, exc_raised) def _suppress_exc(*exc_details): return True def _expect_ok(exc_type, exc, exc_tb): self.assertIsNone(exc_type) self.assertIsNone(exc) self.assertIsNone(exc_tb) class ExitCM(object): def __init__(self, check_exc): self.check_exc = check_exc def __enter__(self): self.fail("Should not be called!") def __exit__(self, *exc_details): self.check_exc(*exc_details) with ExitStack() as stack: stack.push(_expect_ok) self.assertIs(stack._exit_callbacks[-1], _expect_ok) cm = ExitCM(_expect_ok) stack.push(cm) self.assertIs(stack._exit_callbacks[-1].__self__, cm) stack.push(_suppress_exc) self.assertIs(stack._exit_callbacks[-1], _suppress_exc) cm = ExitCM(_expect_exc) stack.push(cm) self.assertIs(stack._exit_callbacks[-1].__self__, cm) stack.push(_expect_exc) self.assertIs(stack._exit_callbacks[-1], _expect_exc) stack.push(_expect_exc) self.assertIs(stack._exit_callbacks[-1], _expect_exc) 1/0 def test_enter_context(self): class TestCM(object): def __enter__(self): result.append(1) def __exit__(self, *exc_details): result.append(3) result = [] cm = TestCM() with ExitStack() as stack: @stack.callback # Registered first => cleaned up last def _exit(): result.append(4) self.assertIsNotNone(_exit) stack.enter_context(cm) self.assertIs(stack._exit_callbacks[-1].__self__, cm) result.append(2) self.assertEqual(result, [1, 2, 3, 4]) def test_close(self): result = [] with ExitStack() as stack: @stack.callback def _exit(): result.append(1) self.assertIsNotNone(_exit) stack.close() result.append(2) self.assertEqual(result, [1, 2]) def test_pop_all(self): result = [] with ExitStack() as stack: @stack.callback def _exit(): result.append(3) self.assertIsNotNone(_exit) new_stack = stack.pop_all() result.append(1) result.append(2) new_stack.close() self.assertEqual(result, [1, 2, 3]) def test_exit_raise(self): with self.assertRaises(ZeroDivisionError): with ExitStack() as stack: stack.push(lambda *exc: False) 1/0 def test_exit_suppress(self): with ExitStack() as stack: stack.push(lambda *exc: True) 1/0 def test_exit_exception_chaining_reference(self): # Sanity check to make sure that ExitStack chaining matches # actual nested with statements class RaiseExc: def __init__(self, exc): self.exc = exc def __enter__(self): return self def __exit__(self, *exc_details): raise self.exc class RaiseExcWithContext: def __init__(self, outer, inner): self.outer = outer self.inner = inner def __enter__(self): return self def __exit__(self, *exc_details): try: raise self.inner except: raise self.outer class SuppressExc: def __enter__(self): return self def __exit__(self, *exc_details): type(self).saved_details = exc_details return True try: with RaiseExc(IndexError): with RaiseExcWithContext(KeyError, AttributeError): with SuppressExc(): with RaiseExc(ValueError): 1 / 0 except IndexError as exc: self.assertIsInstance(exc.__context__, KeyError) self.assertIsInstance(exc.__context__.__context__, AttributeError) # Inner exceptions were suppressed self.assertIsNone(exc.__context__.__context__.__context__) else: self.fail("Expected IndexError, but no exception was raised") # Check the inner exceptions inner_exc = SuppressExc.saved_details[1] self.assertIsInstance(inner_exc, ValueError) self.assertIsInstance(inner_exc.__context__, ZeroDivisionError) def test_exit_exception_chaining(self): # Ensure exception chaining matches the reference behaviour def raise_exc(exc): raise exc saved_details = None def suppress_exc(*exc_details): nonlocal saved_details saved_details = exc_details return True try: with ExitStack() as stack: stack.callback(raise_exc, IndexError) stack.callback(raise_exc, KeyError) stack.callback(raise_exc, AttributeError) stack.push(suppress_exc) stack.callback(raise_exc, ValueError) 1 / 0 except IndexError as exc: self.assertIsInstance(exc.__context__, KeyError) self.assertIsInstance(exc.__context__.__context__, AttributeError) # Inner exceptions were suppressed self.assertIsNone(exc.__context__.__context__.__context__) else: self.fail("Expected IndexError, but no exception was raised") # Check the inner exceptions inner_exc = saved_details[1] self.assertIsInstance(inner_exc, ValueError) self.assertIsInstance(inner_exc.__context__, ZeroDivisionError) def test_exit_exception_non_suppressing(self): # http://bugs.python.org/issue19092 def raise_exc(exc): raise exc def suppress_exc(*exc_details): return True try: with ExitStack() as stack: stack.callback(lambda: None) stack.callback(raise_exc, IndexError) except Exception as exc: self.assertIsInstance(exc, IndexError) else: self.fail("Expected IndexError, but no exception was raised") try: with ExitStack() as stack: stack.callback(raise_exc, KeyError) stack.push(suppress_exc) stack.callback(raise_exc, IndexError) except Exception as exc: self.assertIsInstance(exc, KeyError) else: self.fail("Expected KeyError, but no exception was raised") def test_exit_exception_with_correct_context(self): # http://bugs.python.org/issue20317 @contextmanager def gets_the_context_right(exc): try: yield finally: raise exc exc1 = Exception(1) exc2 = Exception(2) exc3 = Exception(3) exc4 = Exception(4) # The contextmanager already fixes the context, so prior to the # fix, ExitStack would try to fix it *again* and get into an # infinite self-referential loop try: with ExitStack() as stack: stack.enter_context(gets_the_context_right(exc4)) stack.enter_context(gets_the_context_right(exc3)) stack.enter_context(gets_the_context_right(exc2)) raise exc1 except Exception as exc: self.assertIs(exc, exc4) self.assertIs(exc.__context__, exc3) self.assertIs(exc.__context__.__context__, exc2) self.assertIs(exc.__context__.__context__.__context__, exc1) self.assertIsNone( exc.__context__.__context__.__context__.__context__) def test_exit_exception_with_existing_context(self): # Addresses a lack of test coverage discovered after checking in a # fix for issue 20317 that still contained debugging code. def raise_nested(inner_exc, outer_exc): try: raise inner_exc finally: raise outer_exc exc1 = Exception(1) exc2 = Exception(2) exc3 = Exception(3) exc4 = Exception(4) exc5 = Exception(5) try: with ExitStack() as stack: stack.callback(raise_nested, exc4, exc5) stack.callback(raise_nested, exc2, exc3) raise exc1 except Exception as exc: self.assertIs(exc, exc5) self.assertIs(exc.__context__, exc4) self.assertIs(exc.__context__.__context__, exc3) self.assertIs(exc.__context__.__context__.__context__, exc2) self.assertIs( exc.__context__.__context__.__context__.__context__, exc1) self.assertIsNone( exc.__context__.__context__.__context__.__context__.__context__) def test_body_exception_suppress(self): def suppress_exc(*exc_details): return True try: with ExitStack() as stack: stack.push(suppress_exc) 1/0 except IndexError as exc: self.fail("Expected no exception, got IndexError") def test_exit_exception_chaining_suppress(self): with ExitStack() as stack: stack.push(lambda *exc: True) stack.push(lambda *exc: 1/0) stack.push(lambda *exc: {}[1]) def test_excessive_nesting(self): # The original implementation would die with RecursionError here with ExitStack() as stack: for i in range(10000): stack.callback(int) def test_instance_bypass(self): class Example(object): pass cm = Example() cm.__exit__ = object() stack = ExitStack() self.assertRaises(AttributeError, stack.enter_context, cm) stack.push(cm) self.assertIs(stack._exit_callbacks[-1], cm) def test_dont_reraise_RuntimeError(self): # https://bugs.python.org/issue27122 class UniqueException(Exception): pass class UniqueRuntimeError(RuntimeError): pass @contextmanager def second(): try: yield 1 except Exception as exc: raise UniqueException("new exception") from exc @contextmanager def first(): try: yield 1 except Exception as exc: raise exc # The UniqueRuntimeError should be caught by second()'s exception # handler which chain raised a new UniqueException. with self.assertRaises(UniqueException) as err_ctx: with ExitStack() as es_ctx: es_ctx.enter_context(second()) es_ctx.enter_context(first()) raise UniqueRuntimeError("please no infinite loop.") exc = err_ctx.exception self.assertIsInstance(exc, UniqueException) self.assertIsInstance(exc.__context__, UniqueRuntimeError) self.assertIsNone(exc.__context__.__context__) self.assertIsNone(exc.__context__.__cause__) self.assertIs(exc.__cause__, exc.__context__) class TestRedirectStream: redirect_stream = None orig_stream = None @support.requires_docstrings def test_instance_docs(self): # Issue 19330: ensure context manager instances have good docstrings cm_docstring = self.redirect_stream.__doc__ obj = self.redirect_stream(None) self.assertEqual(obj.__doc__, cm_docstring) def test_no_redirect_in_init(self): orig_stdout = getattr(sys, self.orig_stream) self.redirect_stream(None) self.assertIs(getattr(sys, self.orig_stream), orig_stdout) def test_redirect_to_string_io(self): f = io.StringIO() msg = "Consider an API like help(), which prints directly to stdout" orig_stdout = getattr(sys, self.orig_stream) with self.redirect_stream(f): print(msg, file=getattr(sys, self.orig_stream)) self.assertIs(getattr(sys, self.orig_stream), orig_stdout) s = f.getvalue().strip() self.assertEqual(s, msg) def test_enter_result_is_target(self): f = io.StringIO() with self.redirect_stream(f) as enter_result: self.assertIs(enter_result, f) def test_cm_is_reusable(self): f = io.StringIO() write_to_f = self.redirect_stream(f) orig_stdout = getattr(sys, self.orig_stream) with write_to_f: print("Hello", end=" ", file=getattr(sys, self.orig_stream)) with write_to_f: print("World!", file=getattr(sys, self.orig_stream)) self.assertIs(getattr(sys, self.orig_stream), orig_stdout) s = f.getvalue() self.assertEqual(s, "Hello World!\n") def test_cm_is_reentrant(self): f = io.StringIO() write_to_f = self.redirect_stream(f) orig_stdout = getattr(sys, self.orig_stream) with write_to_f: print("Hello", end=" ", file=getattr(sys, self.orig_stream)) with write_to_f: print("World!", file=getattr(sys, self.orig_stream)) self.assertIs(getattr(sys, self.orig_stream), orig_stdout) s = f.getvalue() self.assertEqual(s, "Hello World!\n") class TestRedirectStdout(TestRedirectStream, unittest.TestCase): redirect_stream = redirect_stdout orig_stream = "stdout" class TestRedirectStderr(TestRedirectStream, unittest.TestCase): redirect_stream = redirect_stderr orig_stream = "stderr" class TestSuppress(unittest.TestCase): @support.requires_docstrings def test_instance_docs(self): # Issue 19330: ensure context manager instances have good docstrings cm_docstring = suppress.__doc__ obj = suppress() self.assertEqual(obj.__doc__, cm_docstring) def test_no_result_from_enter(self): with suppress(ValueError) as enter_result: self.assertIsNone(enter_result) def test_no_exception(self): with suppress(ValueError): self.assertEqual(pow(2, 5), 32) def test_exact_exception(self): with suppress(TypeError): len(5) def test_exception_hierarchy(self): with suppress(LookupError): 'Hello'[50] def test_other_exception(self): with self.assertRaises(ZeroDivisionError): with suppress(TypeError): 1/0 def test_no_args(self): with self.assertRaises(ZeroDivisionError): with suppress(): 1/0 def test_multiple_exception_args(self): with suppress(ZeroDivisionError, TypeError): 1/0 with suppress(ZeroDivisionError, TypeError): len(5) def test_cm_is_reentrant(self): ignore_exceptions = suppress(Exception) with ignore_exceptions: pass with ignore_exceptions: len(5) with ignore_exceptions: with ignore_exceptions: # Check nested usage len(5) outer_continued = True 1/0 self.assertTrue(outer_continued) if __name__ == "__main__": unittest.main()