except StopIteration:
return False
else:
- raise RuntimeError("generator didn't stop")
+ try:
+ raise RuntimeError("generator didn't stop")
+ finally:
+ self.gen.close()
else:
if value is None:
# Need to force instantiation so we can reliably
raise
exc.__traceback__ = traceback
return False
- raise RuntimeError("generator didn't stop after throw()")
+ try:
+ raise RuntimeError("generator didn't stop after throw()")
+ finally:
+ self.gen.close()
class _AsyncGeneratorContextManager(
_GeneratorContextManagerBase,
except StopAsyncIteration:
return False
else:
- raise RuntimeError("generator didn't stop")
+ try:
+ raise RuntimeError("generator didn't stop")
+ finally:
+ await self.gen.aclose()
else:
if value is None:
# Need to force instantiation so we can reliably
raise
exc.__traceback__ = traceback
return False
- raise RuntimeError("generator didn't stop after athrow()")
+ try:
+ raise RuntimeError("generator didn't stop after athrow()")
+ finally:
+ await self.gen.aclose()
def contextmanager(func):
yield
ctx = whoo()
ctx.__enter__()
- self.assertRaises(
- RuntimeError, ctx.__exit__, TypeError, TypeError("foo"), None
- )
+ with self.assertRaises(RuntimeError):
+ ctx.__exit__(TypeError, TypeError("foo"), None)
+ if support.check_impl_detail(cpython=True):
+ # The "gen" attribute is an implementation detail.
+ self.assertFalse(ctx.gen.gi_suspended)
+
+ def test_contextmanager_trap_second_yield(self):
+ @contextmanager
+ def whoo():
+ yield
+ yield
+ ctx = whoo()
+ ctx.__enter__()
+ with self.assertRaises(RuntimeError):
+ ctx.__exit__(None, None, None)
+ if support.check_impl_detail(cpython=True):
+ # The "gen" attribute is an implementation detail.
+ self.assertFalse(ctx.gen.gi_suspended)
def test_contextmanager_except(self):
state = []