PEP 343 -- the with-statement.
This was started by Mike Bland and completed by Guido
(with help from Neal).
This still needs a __future__ statement added;
Thomas is working on Michael's patch for that aspect.
There's a small amount of code cleanup and refactoring
in ast.c, compile.c and ceval.c (I fixed the lltrace
behavior when EXT_POP is used -- however I had to make
lltrace a static global).
diff --git a/Lib/compiler/ast.py b/Lib/compiler/ast.py
index accda45..e270995 100644
--- a/Lib/compiler/ast.py
+++ b/Lib/compiler/ast.py
@@ -553,7 +553,7 @@
self.varargs = 1
if flags & CO_VARKEYWORDS:
self.kwargs = 1
-
+
def getChildren(self):
@@ -584,7 +584,7 @@
self.lineno = lineno
self.argnames = ['[outmost-iterable]']
self.varargs = self.kwargs = None
-
+
def getChildren(self):
@@ -763,7 +763,7 @@
self.varargs = 1
if flags & CO_VARKEYWORDS:
self.kwargs = 1
-
+
def getChildren(self):
@@ -1297,6 +1297,31 @@
def __repr__(self):
return "While(%s, %s, %s)" % (repr(self.test), repr(self.body), repr(self.else_))
+class With(Node):
+ def __init__(self, expr, vars, body, lineno=None):
+ self.expr = expr
+ self.vars = vars
+ self.body = body
+ self.lineno = lineno
+
+ def getChildren(self):
+ children = []
+ children.append(self.expr)
+ children.append(self.vars)
+ children.append(self.body)
+ return tuple(children)
+
+ def getChildNodes(self):
+ nodelist = []
+ nodelist.append(self.expr)
+ if self.vars is not None:
+ nodelist.append(self.vars)
+ nodelist.append(self.body)
+ return tuple(nodelist)
+
+ def __repr__(self):
+ return "With(%s, %s, %s)" % (repr(self.expr), repr(self.vars), repr(self.body))
+
class Yield(Node):
def __init__(self, value, lineno=None):
self.value = value
diff --git a/Lib/opcode.py b/Lib/opcode.py
index 9517c43..095ca42 100644
--- a/Lib/opcode.py
+++ b/Lib/opcode.py
@@ -41,6 +41,7 @@
hasjabs.append(op)
# Instruction opcodes for compiled code
+# Blank lines correspond to available opcodes
def_op('STOP_CODE', 0)
def_op('POP_TOP', 1)
@@ -59,7 +60,6 @@
def_op('LIST_APPEND', 18)
def_op('BINARY_POWER', 19)
-
def_op('BINARY_MULTIPLY', 20)
def_op('BINARY_DIVIDE', 21)
def_op('BINARY_MODULO', 22)
@@ -70,7 +70,6 @@
def_op('BINARY_TRUE_DIVIDE', 27)
def_op('INPLACE_FLOOR_DIVIDE', 28)
def_op('INPLACE_TRUE_DIVIDE', 29)
-
def_op('SLICE+0', 30)
def_op('SLICE+1', 31)
def_op('SLICE+2', 32)
@@ -93,7 +92,6 @@
def_op('INPLACE_MODULO', 59)
def_op('STORE_SUBSCR', 60)
def_op('DELETE_SUBSCR', 61)
-
def_op('BINARY_LSHIFT', 62)
def_op('BINARY_RSHIFT', 63)
def_op('BINARY_AND', 64)
@@ -113,13 +111,12 @@
def_op('INPLACE_XOR', 78)
def_op('INPLACE_OR', 79)
def_op('BREAK_LOOP', 80)
-
+def_op('WITH_CLEANUP', 81)
def_op('LOAD_LOCALS', 82)
def_op('RETURN_VALUE', 83)
def_op('IMPORT_STAR', 84)
def_op('EXEC_STMT', 85)
def_op('YIELD_VALUE', 86)
-
def_op('POP_BLOCK', 87)
def_op('END_FINALLY', 88)
def_op('BUILD_CLASS', 89)
@@ -171,7 +168,6 @@
def_op('CALL_FUNCTION', 131) # #args + (#kwargs << 8)
def_op('MAKE_FUNCTION', 132) # Number of args with default values
def_op('BUILD_SLICE', 133) # Number of items
-
def_op('MAKE_CLOSURE', 134)
def_op('LOAD_CLOSURE', 135)
hasfree.append(135)
@@ -183,7 +179,6 @@
def_op('CALL_FUNCTION_VAR', 140) # #args + (#kwargs << 8)
def_op('CALL_FUNCTION_KW', 141) # #args + (#kwargs << 8)
def_op('CALL_FUNCTION_VAR_KW', 142) # #args + (#kwargs << 8)
-
def_op('EXTENDED_ARG', 143)
EXTENDED_ARG = 143
diff --git a/Lib/symbol.py b/Lib/symbol.py
index dd83740..c650138 100755
--- a/Lib/symbol.py
+++ b/Lib/symbol.py
@@ -50,48 +50,50 @@
while_stmt = 293
for_stmt = 294
try_stmt = 295
-except_clause = 296
-suite = 297
-testlist_safe = 298
-old_test = 299
-old_lambdef = 300
-test = 301
-or_test = 302
-and_test = 303
-not_test = 304
-comparison = 305
-comp_op = 306
-expr = 307
-xor_expr = 308
-and_expr = 309
-shift_expr = 310
-arith_expr = 311
-term = 312
-factor = 313
-power = 314
-atom = 315
-listmaker = 316
-testlist_gexp = 317
-lambdef = 318
-trailer = 319
-subscriptlist = 320
-subscript = 321
-sliceop = 322
-exprlist = 323
-testlist = 324
-dictmaker = 325
-classdef = 326
-arglist = 327
-argument = 328
-list_iter = 329
-list_for = 330
-list_if = 331
-gen_iter = 332
-gen_for = 333
-gen_if = 334
-testlist1 = 335
-encoding_decl = 336
-yield_expr = 337
+with_stmt = 296
+with_var = 297
+except_clause = 298
+suite = 299
+testlist_safe = 300
+old_test = 301
+old_lambdef = 302
+test = 303
+or_test = 304
+and_test = 305
+not_test = 306
+comparison = 307
+comp_op = 308
+expr = 309
+xor_expr = 310
+and_expr = 311
+shift_expr = 312
+arith_expr = 313
+term = 314
+factor = 315
+power = 316
+atom = 317
+listmaker = 318
+testlist_gexp = 319
+lambdef = 320
+trailer = 321
+subscriptlist = 322
+subscript = 323
+sliceop = 324
+exprlist = 325
+testlist = 326
+dictmaker = 327
+classdef = 328
+arglist = 329
+argument = 330
+list_iter = 331
+list_for = 332
+list_if = 333
+gen_iter = 334
+gen_for = 335
+gen_if = 336
+testlist1 = 337
+encoding_decl = 338
+yield_expr = 339
#--end constants--
sym_name = {}
diff --git a/Lib/test/contextmanager.py b/Lib/test/contextmanager.py
new file mode 100644
index 0000000..0ebf646
--- /dev/null
+++ b/Lib/test/contextmanager.py
@@ -0,0 +1,34 @@
+class GeneratorContextManager(object):
+ def __init__(self, gen):
+ self.gen = gen
+
+ def __context__(self):
+ return self
+
+ def __enter__(self):
+ try:
+ return self.gen.next()
+ except StopIteration:
+ raise RuntimeError("generator didn't yield")
+
+ def __exit__(self, type, value, traceback):
+ if type is None:
+ try:
+ self.gen.next()
+ except StopIteration:
+ return
+ else:
+ raise RuntimeError("generator didn't stop")
+ else:
+ try:
+ self.gen.throw(type, value, traceback)
+ except (type, StopIteration):
+ return
+ else:
+ raise RuntimeError("generator caught exception")
+
+def contextmanager(func):
+ def helper(*args, **kwds):
+ return GeneratorContextManager(func(*args, **kwds))
+ return helper
+
diff --git a/Lib/test/nested.py b/Lib/test/nested.py
new file mode 100644
index 0000000..bb210d4
--- /dev/null
+++ b/Lib/test/nested.py
@@ -0,0 +1,41 @@
+import sys
+from collections import deque
+
+
+class nested(object):
+ def __init__(self, *contexts):
+ self.contexts = contexts
+ self.entered = None
+
+ def __context__(self):
+ return self
+
+ def __enter__(self):
+ if self.entered is not None:
+ raise RuntimeError("Context is not reentrant")
+ self.entered = deque()
+ vars = []
+ try:
+ for context in self.contexts:
+ mgr = context.__context__()
+ vars.append(mgr.__enter__())
+ self.entered.appendleft(mgr)
+ except:
+ self.__exit__(*sys.exc_info())
+ raise
+ return vars
+
+ def __exit__(self, *exc_info):
+ # Behave like nested with statements
+ # first in, last out
+ # New exceptions override old ones
+ ex = exc_info
+ for mgr in self.entered:
+ try:
+ mgr.__exit__(*ex)
+ except:
+ ex = sys.exc_info()
+ self.entered = None
+ if ex is not exc_info:
+ raise ex[0], ex[1], ex[2]
+
diff --git a/Lib/test/test_with.py b/Lib/test/test_with.py
new file mode 100644
index 0000000..9d809ef
--- /dev/null
+++ b/Lib/test/test_with.py
@@ -0,0 +1,560 @@
+#!/usr/bin/env python
+
+"""Unit tests for the with statement specified in PEP 343."""
+
+__author__ = "Mike Bland"
+__email__ = "mbland at acm dot org"
+
+import unittest
+from test.contextmanager import GeneratorContextManager
+from test.nested import nested
+from test.test_support import run_unittest
+
+
+class MockContextManager(GeneratorContextManager):
+ def __init__(self, gen):
+ GeneratorContextManager.__init__(self, gen)
+ self.context_called = False
+ self.enter_called = False
+ self.exit_called = False
+ self.exit_args = None
+
+ def __context__(self):
+ self.context_called = True
+ return GeneratorContextManager.__context__(self)
+
+ def __enter__(self):
+ self.enter_called = True
+ return GeneratorContextManager.__enter__(self)
+
+ def __exit__(self, type, value, traceback):
+ self.exit_called = True
+ self.exit_args = (type, value, traceback)
+ return GeneratorContextManager.__exit__(self, type, value, traceback)
+
+
+def mock_contextmanager(func):
+ def helper(*args, **kwds):
+ return MockContextManager(func(*args, **kwds))
+ return helper
+
+
+class MockResource(object):
+ def __init__(self):
+ self.yielded = False
+ self.stopped = False
+
+
+@mock_contextmanager
+def mock_contextmanager_generator():
+ mock = MockResource()
+ try:
+ mock.yielded = True
+ yield mock
+ finally:
+ mock.stopped = True
+
+
+class MockNested(nested):
+ def __init__(self, *contexts):
+ nested.__init__(self, *contexts)
+ self.context_called = False
+ self.enter_called = False
+ self.exit_called = False
+ self.exit_args = None
+
+ def __context__(self):
+ self.context_called = True
+ return nested.__context__(self)
+
+ def __enter__(self):
+ self.enter_called = True
+ return nested.__enter__(self)
+
+ def __exit__(self, *exc_info):
+ self.exit_called = True
+ self.exit_args = exc_info
+ return nested.__exit__(self, *exc_info)
+
+
+class FailureTestCase(unittest.TestCase):
+ def testNameError(self):
+ def fooNotDeclared():
+ with foo: pass
+ self.assertRaises(NameError, fooNotDeclared)
+
+ def testContextAttributeError(self):
+ class LacksContext(object):
+ def __enter__(self):
+ pass
+
+ def __exit__(self, type, value, traceback):
+ pass
+
+ def fooLacksContext():
+ foo = LacksContext()
+ with foo: pass
+ self.assertRaises(AttributeError, fooLacksContext)
+
+ def testEnterAttributeError(self):
+ class LacksEnter(object):
+ def __context__(self):
+ pass
+
+ def __exit__(self, type, value, traceback):
+ pass
+
+ def fooLacksEnter():
+ foo = LacksEnter()
+ with foo: pass
+ self.assertRaises(AttributeError, fooLacksEnter)
+
+ def testExitAttributeError(self):
+ class LacksExit(object):
+ def __context__(self):
+ pass
+
+ def __enter__(self):
+ pass
+
+ def fooLacksExit():
+ foo = LacksExit()
+ with foo: pass
+ self.assertRaises(AttributeError, fooLacksExit)
+
+ def assertRaisesSyntaxError(self, codestr):
+ def shouldRaiseSyntaxError(s):
+ compile(s, '', 'single')
+ self.assertRaises(SyntaxError, shouldRaiseSyntaxError, codestr)
+
+ def testAssignmentToNoneError(self):
+ self.assertRaisesSyntaxError('with mock as None:\n pass')
+ self.assertRaisesSyntaxError(
+ 'with mock as (None):\n'
+ ' pass')
+
+ def testAssignmentToEmptyTupleError(self):
+ self.assertRaisesSyntaxError(
+ 'with mock as ():\n'
+ ' pass')
+
+ def testAssignmentToTupleOnlyContainingNoneError(self):
+ self.assertRaisesSyntaxError('with mock as None,:\n pass')
+ self.assertRaisesSyntaxError(
+ 'with mock as (None,):\n'
+ ' pass')
+
+ def testAssignmentToTupleContainingNoneError(self):
+ self.assertRaisesSyntaxError(
+ 'with mock as (foo, None, bar):\n'
+ ' pass')
+
+ def testContextThrows(self):
+ class ContextThrows(object):
+ def __context__(self):
+ raise RuntimeError("Context threw")
+
+ def shouldThrow():
+ ct = ContextThrows()
+ self.foo = None
+ with ct as self.foo:
+ pass
+ self.assertRaises(RuntimeError, shouldThrow)
+ self.assertEqual(self.foo, None)
+
+ def testEnterThrows(self):
+ class EnterThrows(object):
+ def __context__(self):
+ return self
+
+ def __enter__(self):
+ raise RuntimeError("Context threw")
+
+ def __exit__(self, *args):
+ pass
+
+ def shouldThrow():
+ ct = EnterThrows()
+ self.foo = None
+ with ct as self.foo:
+ pass
+ self.assertRaises(RuntimeError, shouldThrow)
+ self.assertEqual(self.foo, None)
+
+ def testExitThrows(self):
+ class ExitThrows(object):
+ def __context__(self):
+ return self
+ def __enter__(self):
+ return
+ def __exit__(self, *args):
+ raise RuntimeError(42)
+ def shouldThrow():
+ with ExitThrows():
+ pass
+ self.assertRaises(RuntimeError, shouldThrow)
+
+class ContextmanagerAssertionMixin(object):
+ TEST_EXCEPTION = RuntimeError("test exception")
+
+ def assertInWithManagerInvariants(self, mock_manager):
+ self.assertTrue(mock_manager.context_called)
+ self.assertTrue(mock_manager.enter_called)
+ self.assertFalse(mock_manager.exit_called)
+ self.assertEqual(mock_manager.exit_args, None)
+
+ def assertAfterWithManagerInvariants(self, mock_manager, exit_args):
+ self.assertTrue(mock_manager.context_called)
+ self.assertTrue(mock_manager.enter_called)
+ self.assertTrue(mock_manager.exit_called)
+ self.assertEqual(mock_manager.exit_args, exit_args)
+
+ def assertAfterWithManagerInvariantsNoError(self, mock_manager):
+ self.assertAfterWithManagerInvariants(mock_manager,
+ (None, None, None))
+
+ def assertInWithGeneratorInvariants(self, mock_generator):
+ self.assertTrue(mock_generator.yielded)
+ self.assertFalse(mock_generator.stopped)
+
+ def assertAfterWithGeneratorInvariantsNoError(self, mock_generator):
+ self.assertTrue(mock_generator.yielded)
+ self.assertTrue(mock_generator.stopped)
+
+ def raiseTestException(self):
+ raise self.TEST_EXCEPTION
+
+ def assertAfterWithManagerInvariantsWithError(self, mock_manager):
+ self.assertTrue(mock_manager.context_called)
+ self.assertTrue(mock_manager.enter_called)
+ self.assertTrue(mock_manager.exit_called)
+ self.assertEqual(mock_manager.exit_args[0], RuntimeError)
+ self.assertEqual(mock_manager.exit_args[1], self.TEST_EXCEPTION)
+
+ def assertAfterWithGeneratorInvariantsWithError(self, mock_generator):
+ self.assertTrue(mock_generator.yielded)
+ self.assertTrue(mock_generator.stopped)
+
+
+class NonexceptionalTestCase(unittest.TestCase, ContextmanagerAssertionMixin):
+ def testInlineGeneratorSyntax(self):
+ with mock_contextmanager_generator():
+ pass
+
+ def testUnboundGenerator(self):
+ mock = mock_contextmanager_generator()
+ with mock:
+ pass
+ self.assertAfterWithManagerInvariantsNoError(mock)
+
+ def testInlineGeneratorBoundSyntax(self):
+ with mock_contextmanager_generator() as foo:
+ self.assertInWithGeneratorInvariants(foo)
+ # FIXME: In the future, we'll try to keep the bound names from leaking
+ self.assertAfterWithGeneratorInvariantsNoError(foo)
+
+ def testInlineGeneratorBoundToExistingVariable(self):
+ foo = None
+ with mock_contextmanager_generator() as foo:
+ self.assertInWithGeneratorInvariants(foo)
+ self.assertAfterWithGeneratorInvariantsNoError(foo)
+
+ def testInlineGeneratorBoundToDottedVariable(self):
+ with mock_contextmanager_generator() as self.foo:
+ self.assertInWithGeneratorInvariants(self.foo)
+ self.assertAfterWithGeneratorInvariantsNoError(self.foo)
+
+ def testBoundGenerator(self):
+ mock = mock_contextmanager_generator()
+ with mock as foo:
+ self.assertInWithGeneratorInvariants(foo)
+ self.assertInWithManagerInvariants(mock)
+ self.assertAfterWithGeneratorInvariantsNoError(foo)
+ self.assertAfterWithManagerInvariantsNoError(mock)
+
+ def testNestedSingleStatements(self):
+ mock_a = mock_contextmanager_generator()
+ with mock_a as foo:
+ mock_b = mock_contextmanager_generator()
+ with mock_b as bar:
+ self.assertInWithManagerInvariants(mock_a)
+ self.assertInWithManagerInvariants(mock_b)
+ self.assertInWithGeneratorInvariants(foo)
+ self.assertInWithGeneratorInvariants(bar)
+ self.assertAfterWithManagerInvariantsNoError(mock_b)
+ self.assertAfterWithGeneratorInvariantsNoError(bar)
+ self.assertInWithManagerInvariants(mock_a)
+ self.assertInWithGeneratorInvariants(foo)
+ self.assertAfterWithManagerInvariantsNoError(mock_a)
+ self.assertAfterWithGeneratorInvariantsNoError(foo)
+
+
+class NestedNonexceptionalTestCase(unittest.TestCase,
+ ContextmanagerAssertionMixin):
+ def testSingleArgInlineGeneratorSyntax(self):
+ with nested(mock_contextmanager_generator()):
+ pass
+
+ def testSingleArgUnbound(self):
+ mock_contextmanager = mock_contextmanager_generator()
+ mock_nested = MockNested(mock_contextmanager)
+ with mock_nested:
+ self.assertInWithManagerInvariants(mock_contextmanager)
+ self.assertInWithManagerInvariants(mock_nested)
+ self.assertAfterWithManagerInvariantsNoError(mock_contextmanager)
+ self.assertAfterWithManagerInvariantsNoError(mock_nested)
+
+ def testSingleArgBoundToNonTuple(self):
+ m = mock_contextmanager_generator()
+ # This will bind all the arguments to nested() into a single list
+ # assigned to foo.
+ with nested(m) as foo:
+ self.assertInWithManagerInvariants(m)
+ self.assertAfterWithManagerInvariantsNoError(m)
+
+ def testSingleArgBoundToSingleElementParenthesizedList(self):
+ m = mock_contextmanager_generator()
+ # This will bind all the arguments to nested() into a single list
+ # assigned to foo.
+ # FIXME: what should this do: with nested(m) as (foo,):
+ with nested(m) as (foo):
+ self.assertInWithManagerInvariants(m)
+ self.assertAfterWithManagerInvariantsNoError(m)
+
+ def testSingleArgBoundToMultipleElementTupleError(self):
+ def shouldThrowValueError():
+ with nested(mock_contextmanager_generator()) as (foo, bar):
+ pass
+ self.assertRaises(ValueError, shouldThrowValueError)
+
+ def testSingleArgUnbound(self):
+ mock_contextmanager = mock_contextmanager_generator()
+ mock_nested = MockNested(mock_contextmanager)
+ with mock_nested:
+ self.assertInWithManagerInvariants(mock_contextmanager)
+ self.assertInWithManagerInvariants(mock_nested)
+ self.assertAfterWithManagerInvariantsNoError(mock_contextmanager)
+ self.assertAfterWithManagerInvariantsNoError(mock_nested)
+
+ def testMultipleArgUnbound(self):
+ m = mock_contextmanager_generator()
+ n = mock_contextmanager_generator()
+ o = mock_contextmanager_generator()
+ mock_nested = MockNested(m, n, o)
+ with mock_nested:
+ self.assertInWithManagerInvariants(m)
+ self.assertInWithManagerInvariants(n)
+ self.assertInWithManagerInvariants(o)
+ self.assertInWithManagerInvariants(mock_nested)
+ self.assertAfterWithManagerInvariantsNoError(m)
+ self.assertAfterWithManagerInvariantsNoError(n)
+ self.assertAfterWithManagerInvariantsNoError(o)
+ self.assertAfterWithManagerInvariantsNoError(mock_nested)
+
+ def testMultipleArgBound(self):
+ mock_nested = MockNested(mock_contextmanager_generator(),
+ mock_contextmanager_generator(), mock_contextmanager_generator())
+ with mock_nested as (m, n, o):
+ self.assertInWithGeneratorInvariants(m)
+ self.assertInWithGeneratorInvariants(n)
+ self.assertInWithGeneratorInvariants(o)
+ self.assertInWithManagerInvariants(mock_nested)
+ self.assertAfterWithGeneratorInvariantsNoError(m)
+ self.assertAfterWithGeneratorInvariantsNoError(n)
+ self.assertAfterWithGeneratorInvariantsNoError(o)
+ self.assertAfterWithManagerInvariantsNoError(mock_nested)
+
+
+class ExceptionalTestCase(unittest.TestCase, ContextmanagerAssertionMixin):
+ def testSingleResource(self):
+ cm = mock_contextmanager_generator()
+ def shouldThrow():
+ with cm as self.resource:
+ self.assertInWithManagerInvariants(cm)
+ self.assertInWithGeneratorInvariants(self.resource)
+ self.raiseTestException()
+ self.assertRaises(RuntimeError, shouldThrow)
+ self.assertAfterWithManagerInvariantsWithError(cm)
+ self.assertAfterWithGeneratorInvariantsWithError(self.resource)
+
+ def testNestedSingleStatements(self):
+ mock_a = mock_contextmanager_generator()
+ mock_b = mock_contextmanager_generator()
+ def shouldThrow():
+ with mock_a as self.foo:
+ with mock_b as self.bar:
+ self.assertInWithManagerInvariants(mock_a)
+ self.assertInWithManagerInvariants(mock_b)
+ self.assertInWithGeneratorInvariants(self.foo)
+ self.assertInWithGeneratorInvariants(self.bar)
+ self.raiseTestException()
+ self.assertRaises(RuntimeError, shouldThrow)
+ self.assertAfterWithManagerInvariantsWithError(mock_a)
+ self.assertAfterWithManagerInvariantsWithError(mock_b)
+ self.assertAfterWithGeneratorInvariantsWithError(self.foo)
+ self.assertAfterWithGeneratorInvariantsWithError(self.bar)
+
+ def testMultipleResourcesInSingleStatement(self):
+ cm_a = mock_contextmanager_generator()
+ cm_b = mock_contextmanager_generator()
+ mock_nested = MockNested(cm_a, cm_b)
+ def shouldThrow():
+ with mock_nested as (self.resource_a, self.resource_b):
+ self.assertInWithManagerInvariants(cm_a)
+ self.assertInWithManagerInvariants(cm_b)
+ self.assertInWithManagerInvariants(mock_nested)
+ self.assertInWithGeneratorInvariants(self.resource_a)
+ self.assertInWithGeneratorInvariants(self.resource_b)
+ self.raiseTestException()
+ self.assertRaises(RuntimeError, shouldThrow)
+ self.assertAfterWithManagerInvariantsWithError(cm_a)
+ self.assertAfterWithManagerInvariantsWithError(cm_b)
+ self.assertAfterWithManagerInvariantsWithError(mock_nested)
+ self.assertAfterWithGeneratorInvariantsWithError(self.resource_a)
+ self.assertAfterWithGeneratorInvariantsWithError(self.resource_b)
+
+ def testNestedExceptionBeforeInnerStatement(self):
+ mock_a = mock_contextmanager_generator()
+ mock_b = mock_contextmanager_generator()
+ self.bar = None
+ def shouldThrow():
+ with mock_a as self.foo:
+ self.assertInWithManagerInvariants(mock_a)
+ self.assertInWithGeneratorInvariants(self.foo)
+ self.raiseTestException()
+ with mock_b as self.bar:
+ pass
+ self.assertRaises(RuntimeError, shouldThrow)
+ self.assertAfterWithManagerInvariantsWithError(mock_a)
+ self.assertAfterWithGeneratorInvariantsWithError(self.foo)
+
+ # The inner statement stuff should never have been touched
+ self.assertEqual(self.bar, None)
+ self.assertFalse(mock_b.context_called)
+ self.assertFalse(mock_b.enter_called)
+ self.assertFalse(mock_b.exit_called)
+ self.assertEqual(mock_b.exit_args, None)
+
+ def testNestedExceptionAfterInnerStatement(self):
+ mock_a = mock_contextmanager_generator()
+ mock_b = mock_contextmanager_generator()
+ def shouldThrow():
+ with mock_a as self.foo:
+ with mock_b as self.bar:
+ self.assertInWithManagerInvariants(mock_a)
+ self.assertInWithManagerInvariants(mock_b)
+ self.assertInWithGeneratorInvariants(self.foo)
+ self.assertInWithGeneratorInvariants(self.bar)
+ self.raiseTestException()
+ self.assertRaises(RuntimeError, shouldThrow)
+ self.assertAfterWithManagerInvariantsWithError(mock_a)
+ self.assertAfterWithManagerInvariantsNoError(mock_b)
+ self.assertAfterWithGeneratorInvariantsWithError(self.foo)
+ self.assertAfterWithGeneratorInvariantsNoError(self.bar)
+
+
+class NonLocalFlowControlTestCase(unittest.TestCase):
+
+ def testWithBreak(self):
+ counter = 0
+ while True:
+ counter += 1
+ with mock_contextmanager_generator():
+ counter += 10
+ break
+ counter += 100 # Not reached
+ self.assertEqual(counter, 11)
+
+ def testWithContinue(self):
+ counter = 0
+ while True:
+ counter += 1
+ if counter > 2:
+ break
+ with mock_contextmanager_generator():
+ counter += 10
+ continue
+ counter += 100 # Not reached
+ self.assertEqual(counter, 12)
+
+ def testWithReturn(self):
+ def foo():
+ counter = 0
+ while True:
+ counter += 1
+ with mock_contextmanager_generator():
+ counter += 10
+ return counter
+ counter += 100 # Not reached
+ self.assertEqual(foo(), 11)
+
+ def testWithYield(self):
+ def gen():
+ with mock_contextmanager_generator():
+ yield 12
+ yield 13
+ x = list(gen())
+ self.assertEqual(x, [12, 13])
+
+ def testWithRaise(self):
+ counter = 0
+ try:
+ counter += 1
+ with mock_contextmanager_generator():
+ counter += 10
+ raise RuntimeError
+ counter += 100 # Not reached
+ except RuntimeError:
+ self.assertEqual(counter, 11)
+ else:
+ self.fail("Didn't raise RuntimeError")
+
+
+class AssignmentTargetTestCase(unittest.TestCase):
+
+ def testSingleComplexTarget(self):
+ targets = {1: [0, 1, 2]}
+ with mock_contextmanager_generator() as targets[1][0]:
+ self.assertEqual(targets.keys(), [1])
+ self.assertEqual(targets[1][0].__class__, MockResource)
+ with mock_contextmanager_generator() as targets.values()[0][1]:
+ self.assertEqual(targets.keys(), [1])
+ self.assertEqual(targets[1][1].__class__, MockResource)
+ with mock_contextmanager_generator() as targets[2]:
+ keys = targets.keys()
+ keys.sort()
+ self.assertEqual(keys, [1, 2])
+ class C: pass
+ blah = C()
+ with mock_contextmanager_generator() as blah.foo:
+ self.assertEqual(hasattr(blah, "foo"), True)
+
+ def testMultipleComplexTargets(self):
+ class C:
+ def __context__(self): return self
+ def __enter__(self): return 1, 2, 3
+ def __exit__(self, *a): pass
+ targets = {1: [0, 1, 2]}
+ with C() as (targets[1][0], targets[1][1], targets[1][2]):
+ self.assertEqual(targets, {1: [1, 2, 3]})
+ with C() as (targets.values()[0][2], targets.values()[0][1], targets.values()[0][0]):
+ self.assertEqual(targets, {1: [3, 2, 1]})
+ with C() as (targets[1], targets[2], targets[3]):
+ self.assertEqual(targets, {1: 1, 2: 2, 3: 3})
+ class B: pass
+ blah = B()
+ with C() as (blah.one, blah.two, blah.three):
+ self.assertEqual(blah.one, 1)
+ self.assertEqual(blah.two, 2)
+ self.assertEqual(blah.three, 3)
+
+
+def test_main():
+ run_unittest(FailureTestCase, NonexceptionalTestCase,
+ NestedNonexceptionalTestCase, ExceptionalTestCase,
+ NonLocalFlowControlTestCase,
+ AssignmentTargetTestCase)
+
+
+if __name__ == '__main__':
+ test_main()