# Running after the first yield
next(self.generator)
+ def test_types_coroutine_wrapper_state(self):
+ def gen():
+ yield 1
+ yield 2
+
+ @types.coroutine
+ def wrapped_generator_coro():
+ # return a generator iterator so types.coroutine
+ # wraps it into types._GeneratorWrapper.
+ return gen()
+
+ g = wrapped_generator_coro()
+ self.addCleanup(g.close)
+ self.assertIs(type(g), types._GeneratorWrapper)
+
+ # _GeneratorWrapper must provide gi_suspended/cr_suspended
+ # so inspect.get*state() doesn't raise AttributeError.
+ self.assertEqual(inspect.getgeneratorstate(g), inspect.GEN_CREATED)
+ self.assertEqual(inspect.getcoroutinestate(g), inspect.CORO_CREATED)
+
+ next(g)
+ self.assertEqual(inspect.getgeneratorstate(g), inspect.GEN_SUSPENDED)
+ self.assertEqual(inspect.getcoroutinestate(g), inspect.CORO_SUSPENDED)
+
def test_easy_debugging(self):
# repr() and str() of a generator state should contain the state name
names = 'GEN_CREATED GEN_RUNNING GEN_SUSPENDED GEN_CLOSED'.split()
@property
def gi_yieldfrom(self):
return self.__wrapped.gi_yieldfrom
+ @property
+ def gi_suspended(self):
+ return self.__wrapped.gi_suspended
cr_code = gi_code
cr_frame = gi_frame
cr_running = gi_running
cr_await = gi_yieldfrom
+ cr_suspended = gi_suspended
def __next__(self):
return next(self.__wrapped)
def __iter__(self):