import copy

import gc

import pickle

import sys

import doctest

import unittest

import weakref

import inspect

import textwrap

import types

from test import support

try:

import _testcapi

except ImportError:

_testcapi = None

# This tests to make sure that if a SIGINT arrives just before we send into a

# yield from chain, the KeyboardInterrupt is raised in the innermost

# generator (see bpo-30039).

@unittest.skipUnless(_testcapi is not None and

hasattr(_testcapi, "raise_SIGINT_then_send_None"),

"needs _testcapi.raise_SIGINT_then_send_None")

class SignalAndYieldFromTest(unittest.TestCase):

def generator1(self):

return (yield from self.generator2())

def generator2(self):

try:

yield

except KeyboardInterrupt:

return "PASSED"

else:

return "FAILED"

def test_raise_and_yield_from(self):

gen = self.generator1()

gen.send(None)

try:

_testcapi.raise_SIGINT_then_send_None(gen)

except BaseException as _exc:

exc = _exc

self.assertIs(type(exc), StopIteration)

self.assertEqual(exc.value, "PASSED")

class FinalizationTest(unittest.TestCase):

def test_frame_resurrect(self):

# A generator frame can be resurrected by a generator's finalization.

def gen():

nonlocal frame

try:

yield

finally:

frame = sys._getframe()

g = gen()

wr = weakref.ref(g)

next(g)

del g

support.gc_collect()

self.assertIs(wr(), None)

self.assertTrue(frame)

del frame

support.gc_collect()

def test_refcycle(self):

# A generator caught in a refcycle gets finalized anyway.

old_garbage = gc.garbage[:]

finalized = False

def gen():

nonlocal finalized

try:

g = yield

yield 1

finally:

finalized = True

g = gen()

next(g)

g.send(g)

self.assertGreaterEqual(sys.getrefcount(g), 2)

self.assertFalse(finalized)

del g

support.gc_collect()

self.assertTrue(finalized)

self.assertEqual(gc.garbage, old_garbage)

def test_lambda_generator(self):

# bpo-23192, gh-119897: Test that a lambda returning a generator behaves

# like the equivalent function

f = lambda: (yield 1)

self.assertIsInstance(f(), types.GeneratorType)

self.assertEqual(next(f()), 1)

def g(): return (yield 1)

# test 'yield from'

f2 = lambda: (yield from g())

def g2(): return (yield from g())

f3 = lambda: (yield from f())

def g3(): return (yield from f())

for gen_fun in (f, g, f2, g2, f3, g3):

gen = gen_fun()

self.assertEqual(next(gen), 1)

with self.assertRaises(StopIteration) as cm:

gen.send(2)

self.assertEqual(cm.exception.value, 2)

def test_generator_resurrect(self):

# Test that a resurrected generator still has a valid gi_code

resurrected = []

# Resurrect a generator in a finalizer

exec(textwrap.dedent("""

def gen():

try:

yield

except:

resurrected.append(g)

g = gen()

next(g)

"""), {"resurrected": resurrected})

support.gc_collect()

self.assertEqual(len(resurrected), 1)

self.assertIsInstance(resurrected[0].gi_code, types.CodeType)

def test_exhausted_generator_frame_cycle(self):

def g():

yield

generator = g()

frame = generator.gi_frame

self.assertIsNone(frame.f_back)

next(generator)

self.assertIsNone(frame.f_back)

next(generator, None)

self.assertIsNone(frame.f_back)

class GeneratorTest(unittest.TestCase):

def test_name(self):

def func():

yield 1

# check generator names

gen = func()

self.assertEqual(gen.__name__, "func")

self.assertEqual(gen.__qualname__,

"GeneratorTest.test_name.<locals>.func")

# modify generator names

gen.__name__ = "name"

gen.__qualname__ = "qualname"

self.assertEqual(gen.__name__, "name")

self.assertEqual(gen.__qualname__, "qualname")

# generator names must be a string and cannot be deleted

self.assertRaises(TypeError, setattr, gen, '__name__', 123)

self.assertRaises(TypeError, setattr, gen, '__qualname__', 123)

self.assertRaises(TypeError, delattr, gen, '__name__')

self.assertRaises(TypeError, delattr, gen, '__qualname__')

# modify names of the function creating the generator

func.__qualname__ = "func_qualname"

func.__name__ = "func_name"

gen = func()

self.assertEqual(gen.__name__, "func_name")

self.assertEqual(gen.__qualname__, "func_qualname")

# unnamed generator

gen = (x for x in range(10))

self.assertEqual(gen.__name__,

"<genexpr>")

self.assertEqual(gen.__qualname__,

"GeneratorTest.test_name.<locals>.<genexpr>")

def test_copy(self):

def f():

yield 1

g = f()

with self.assertRaises(TypeError):

copy.copy(g)

def test_pickle(self):

def f():

yield 1

g = f()

for proto in range(pickle.HIGHEST_PROTOCOL + 1):

with self.assertRaises((TypeError, pickle.PicklingError)):

pickle.dumps(g, proto)

def test_send_non_none_to_new_gen(self):

def f():

yield 1

g = f()

with self.assertRaises(TypeError):

g.send(0)

self.assertEqual(next(g), 1)

def test_handle_frame_object_in_creation(self):

#Attempt to expose partially constructed frames

#See https://github.com/python/cpython/issues/94262

def cb(*args):

inspect.stack()

def gen():

yield 1

thresholds = gc.get_threshold()

gc.callbacks.append(cb)

gc.set_threshold(1, 0, 0)

try:

gen()

finally:

gc.set_threshold(*thresholds)

gc.callbacks.pop()

class Sneaky:

def __del__(self):

inspect.stack()

sneaky = Sneaky()

sneaky._s = Sneaky()

sneaky._s._s = sneaky

gc.set_threshold(1, 0, 0)

try:

del sneaky

gen()

finally:

gc.set_threshold(*thresholds)

def test_ag_frame_f_back(self):

async def f():

yield

ag = f()

self.assertIsNone(ag.ag_frame.f_back)

def test_cr_frame_f_back(self):

async def f():

pass

cr = f()

self.assertIsNone(cr.cr_frame.f_back)

cr.close() # Suppress RuntimeWarning.

def test_gi_frame_f_back(self):

def f():

yield

gi = f()

self.assertIsNone(gi.gi_frame.f_back)

def test_issue103488(self):

def gen_raises():

yield

raise ValueError()

def loop():

try:

for _ in gen_raises():

if True is False:

return

except ValueError:

pass

#This should not raise

loop()

def test_genexpr_only_calls_dunder_iter_once(self):

class Iterator:

def __init__(self):

self.val = 0

def __next__(self):

if self.val == 2:

raise StopIteration

self.val += 1

return self.val

# No __iter__ method

class C:

def __iter__(self):

return Iterator()

self.assertEqual([1, 2], list(i for i in C()))

def test_close_clears_frame(self):

# gh-142766: Test that closing a generator clears its frame

class DetectDelete:

def __init__(self):

DetectDelete.deleted = False

def __del__(self):

DetectDelete.deleted = True

def generator(arg):

yield

# Test a freshly created generator (not suspended)

g = generator(DetectDelete())

g.close()

self.assertTrue(DetectDelete.deleted)

# Test a suspended generator

g = generator(DetectDelete())

next(g)

g.close()

self.assertTrue(DetectDelete.deleted)

# Clear via gi_frame.clear()

g = generator(DetectDelete())

g.gi_frame.clear()

self.assertTrue(DetectDelete.deleted)

class ModifyUnderlyingIterableTest(unittest.TestCase):

iterables = [

range(0),

range(20),

[1, 2, 3],

(2,),

{13, 48, 211},

frozenset((15, 8, 6)),

{1: 2, 3: 4},

]

non_iterables = [

None,

42,

3.0,

2j,

]

def genexpr(self):

return (x for x in range(10))

def genfunc(self):

def gen(it):

for x in it:

yield x

return gen(range(10))

def process_tests(self, get_generator, is_expr):

err_iterator = "'.*' object is not an iterator"

err_iterable = "'.*' object is not iterable"

for obj in self.iterables:

g_obj = get_generator(obj)

with self.subTest(g_obj=g_obj, obj=obj):

if is_expr:

self.assertRaisesRegex(TypeError, err_iterator, list, g_obj)

else:

self.assertListEqual(list(g_obj), list(obj))

g_iter = get_generator(iter(obj))

with self.subTest(g_iter=g_iter, obj=obj):

self.assertListEqual(list(g_iter), list(obj))

for obj in self.non_iterables:

g_obj = get_generator(obj)

with self.subTest(g_obj=g_obj):

err = err_iterator if is_expr else err_iterable

self.assertRaisesRegex(TypeError, err, list, g_obj)

def test_modify_f_locals(self):

def modify_f_locals(g, local, obj):

g.gi_frame.f_locals[local] = obj

return g

def get_generator_genexpr(obj):

return modify_f_locals(self.genexpr(), '.0', obj)

def get_generator_genfunc(obj):

return modify_f_locals(self.genfunc(), 'it', obj)

self.process_tests(get_generator_genexpr, True)

self.process_tests(get_generator_genfunc, False)

def test_new_gen_from_gi_code(self):

def new_gen_from_gi_code(g, obj):

generator_func = types.FunctionType(g.gi_code, {})

return generator_func(obj)

def get_generator_genexpr(obj):

return new_gen_from_gi_code(self.genexpr(), obj)

def get_generator_genfunc(obj):

return new_gen_from_gi_code(self.genfunc(), obj)

self.process_tests(get_generator_genexpr, True)

self.process_tests(get_generator_genfunc, False)

class ExceptionTest(unittest.TestCase):

# Tests for the issue #23353: check that the currently handled exception

# is correctly saved/restored in PyEval_EvalFrameEx().

def test_except_throw(self):

def store_raise_exc_generator():

try:

self.assertIsNone(sys.exception())

yield

except Exception as exc:

# exception raised by gen.throw(exc)

self.assertIsInstance(sys.exception(), ValueError)

self.assertIsNone(exc.__context__)

yield

# ensure that the exception is not lost

self.assertIsInstance(sys.exception(), ValueError)

yield

# we should be able to raise back the ValueError

raise

make = store_raise_exc_generator()

next(make)

try:

raise ValueError()

except Exception as exc:

try:

make.throw(exc)

except Exception:

pass

next(make)

with self.assertRaises(ValueError) as cm:

next(make)

self.assertIsNone(cm.exception.__context__)

self.assertIsNone(sys.exception())

def test_except_next(self):

def gen():

self.assertIsInstance(sys.exception(), ValueError)

yield "done"

g = gen()

try:

raise ValueError

except Exception:

self.assertEqual(next(g), "done")

self.assertIsNone(sys.exception())

def test_except_gen_except(self):

def gen():

try:

self.assertIsNone(sys.exception())

yield

# we are called from "except ValueError:", TypeError must

# inherit ValueError in its context

raise TypeError()

except TypeError as exc:

self.assertIsInstance(sys.exception(), TypeError)

self.assertEqual(type(exc.__context__), ValueError)

# here we are still called from the "except ValueError:"

self.assertIsInstance(sys.exception(), ValueError)

yield

self.assertIsNone(sys.exception())

yield "done"

g = gen()

next(g)

try:

raise ValueError

except Exception:

next(g)

self.assertEqual(next(g), "done")

self.assertIsNone(sys.exception())

def test_nested_gen_except_loop(self):

def gen():

for i in range(100):

self.assertIsInstance(sys.exception(), TypeError)

yield "doing"

def outer():

try:

raise TypeError

except:

for x in gen():

yield x

try:

raise ValueError

except Exception:

for x in outer():

self.assertEqual(x, "doing")

self.assertEqual(sys.exception(), None)

def test_except_throw_exception_context(self):

def gen():

try:

try:

self.assertIsNone(sys.exception())

yield

except ValueError:

# we are called from "except ValueError:"

self.assertIsInstance(sys.exception(), ValueError)

raise TypeError()

except Exception as exc:

self.assertIsInstance(sys.exception(), TypeError)

self.assertEqual(type(exc.__context__), ValueError)

# we are still called from "except ValueError:"

self.assertIsInstance(sys.exception(), ValueError)

yield

self.assertIsNone(sys.exception())

yield "done"

g = gen()

next(g)

try:

raise ValueError

except Exception as exc:

g.throw(exc)

self.assertEqual(next(g), "done")

self.assertIsNone(sys.exception())

def test_except_throw_bad_exception(self):

class E(Exception):

def __new__(cls, *args, **kwargs):

return cls

def boring_generator():

yield

gen = boring_generator()

err_msg = 'should have returned an instance of BaseException'

with self.assertRaisesRegex(TypeError, err_msg):

gen.throw(E)

self.assertRaises(StopIteration, next, gen)

def generator():

with self.assertRaisesRegex(TypeError, err_msg):

yield

gen = generator()

next(gen)

with self.assertRaises(StopIteration):

gen.throw(E)

def test_gen_3_arg_deprecation_warning(self):

def g():

yield 42

gen = g()

with self.assertWarns(DeprecationWarning):

with self.assertRaises(TypeError):

gen.throw(TypeError, TypeError(24), None)

def test_stopiteration_error(self):

# See also PEP 479.

def gen():

raise StopIteration

yield

with self.assertRaisesRegex(RuntimeError, 'raised StopIteration'):

next(gen())

def test_tutorial_stopiteration(self):

# Raise StopIteration" stops the generator too:

def f():

yield 1

raise StopIteration

yield 2 # never reached

g = f()

self.assertEqual(next(g), 1)

with self.assertRaisesRegex(RuntimeError, 'raised StopIteration'):

next(g)

def test_return_tuple(self):

def g():

return (yield 1)

gen = g()

self.assertEqual(next(gen), 1)

with self.assertRaises(StopIteration) as cm:

gen.send((2,))

self.assertEqual(cm.exception.value, (2,))

def test_return_stopiteration(self):

def g():

return (yield 1)

gen = g()

self.assertEqual(next(gen), 1)

with self.assertRaises(StopIteration) as cm:

gen.send(StopIteration(2))

self.assertIsInstance(cm.exception.value, StopIteration)

self.assertEqual(cm.exception.value.value, 2)

class GeneratorCloseTest(unittest.TestCase):

def test_close_no_return_value(self):

def f():

yield

gen = f()

gen.send(None)

self.assertIsNone(gen.close())

def test_close_return_value(self):

def f():

try:

yield

# close() raises GeneratorExit here, which is caught

except GeneratorExit:

return 0

gen = f()

gen.send(None)

self.assertEqual(gen.close(), 0)

def test_close_not_catching_exit(self):

def f():

yield

# close() raises GeneratorExit here, which isn't caught and

# therefore propagates -- no return value

return 0

gen = f()

gen.send(None)

self.assertIsNone(gen.close())

def test_close_not_started(self):

def f():

try:

yield

except GeneratorExit:

return 0

gen = f()

self.assertIsNone(gen.close())

def test_close_exhausted(self):

def f():

try:

yield

except GeneratorExit:

return 0

gen = f()

next(gen)

with self.assertRaises(StopIteration):

next(gen)

self.assertIsNone(gen.close())

def test_close_closed(self):

def f():

try:

yield

except GeneratorExit:

return 0

gen = f()

gen.send(None)

self.assertEqual(gen.close(), 0)

self.assertIsNone(gen.close())

def test_close_raises(self):

def f():

try:

yield

except GeneratorExit:

pass

raise RuntimeError

gen = f()

gen.send(None)

with self.assertRaises(RuntimeError):

gen.close()

def test_close_releases_frame_locals(self):

# See gh-118272

class Foo:

pass

f = Foo()

f_wr = weakref.ref(f)

def genfn():

a = f

yield

g = genfn()

next(g)

del f

g.close()

support.gc_collect()

self.assertIsNone(f_wr())

# See https://github.com/python/cpython/issues/125723

class GeneratorDeallocTest(unittest.TestCase):

def test_frame_outlives_generator(self):

def g1():

a = 42

yield sys._getframe()

def g2():

a = 42

yield

def g3(obj):

a = 42

obj.frame = sys._getframe()

yield

class ObjectWithFrame():

def __init__(self):

self.frame = None

def get_frame(index):

if index == 1:

return next(g1())

elif index == 2:

gen = g2()

next(gen)

return gen.gi_frame

elif index == 3:

obj = ObjectWithFrame()

next(g3(obj))

return obj.frame

else:

return None

for index in (1, 2, 3):

with self.subTest(index=index):

frame = get_frame(index)

frame_locals = frame.f_locals

self.assertIn('a', frame_locals)

self.assertEqual(frame_locals['a'], 42)

def test_frame_locals_outlive_generator(self):

frame_locals1 = None

def g1():

nonlocal frame_locals1

frame_locals1 = sys._getframe().f_locals

a = 42

yield

def g2():

a = 42

yield sys._getframe().f_locals

def get_frame_locals(index):

if index == 1:

nonlocal frame_locals1

next(g1())

return frame_locals1

if index == 2:

return next(g2())

else:

return None

for index in (1, 2):

with self.subTest(index=index):

frame_locals = get_frame_locals(index)

self.assertIn('a', frame_locals)

self.assertEqual(frame_locals['a'], 42)

def test_frame_locals_outlive_generator_with_exec(self):

def g():

a = 42

yield locals(), sys._getframe().f_locals

locals_ = {'g': g}

for i in range(10):

exec("snapshot, live_locals = next(g())", locals=locals_)

for l in (locals_['snapshot'], locals_['live_locals']):

self.assertIn('a', l)

self.assertEqual(l['a'], 42)

class GeneratorThrowTest(unittest.TestCase):

def test_exception_context_with_yield(self):

def f():

try:

raise KeyError('a')

except Exception:

yield

gen = f()

gen.send(None)

with self.assertRaises(ValueError) as cm:

gen.throw(ValueError)

context = cm.exception.__context__

self.assertEqual((type(context), context.args), (KeyError, ('a',)))

def test_exception_context_with_yield_inside_generator(self):

# Check that the context is also available from inside the generator

# with yield, as opposed to outside.

def f():

try:

raise KeyError('a')

except Exception:

try:

yield

except Exception as exc:

self.assertEqual(type(exc), ValueError)

context = exc.__context__

self.assertEqual((type(context), context.args),

(KeyError, ('a',)))

yield 'b'

gen = f()

gen.send(None)

actual = gen.throw(ValueError)

# This ensures that the assertions inside were executed.

self.assertEqual(actual, 'b')

def test_exception_context_with_yield_from(self):

def f():

yield

def g():

try:

raise KeyError('a')

except Exception:

yield from f()

gen = g()

gen.send(None)

with self.assertRaises(ValueError) as cm:

gen.throw(ValueError)

context = cm.exception.__context__

self.assertEqual((type(context), context.args), (KeyError, ('a',)))

def test_exception_context_with_yield_from_with_context_cycle(self):

# Check trying to create an exception context cycle:

# https://bugs.python.org/issue40696

has_cycle = None

def f():

yield

def g(exc):

nonlocal has_cycle

try:

raise exc

except Exception:

try:

yield from f()

except Exception as exc:

has_cycle = (exc is exc.__context__)

yield

exc = KeyError('a')

gen = g(exc)

gen.send(None)

gen.throw(exc)

# This also distinguishes from the initial has_cycle=None.

self.assertEqual(has_cycle, False)

def test_throw_after_none_exc_type(self):

def g():

try:

raise KeyError

except KeyError:

pass

try:

yield

except Exception:

raise RuntimeError

gen = g()

gen.send(None)

with self.assertRaises(RuntimeError) as cm:

gen.throw(ValueError)

class GeneratorStackTraceTest(unittest.TestCase):

def check_stack_names(self, frame, expected):

names = []

while frame:

name = frame.f_code.co_name

# Stop checking frames when we get to our test helper.

if (name.startswith('check_') or name.startswith('call_')

or name.startswith('test')):

break

names.append(name)

frame = frame.f_back

self.assertEqual(names, expected)

def check_yield_from_example(self, call_method):

def f():

self.check_stack_names(sys._getframe(), ['f', 'g'])

try:

yield

except Exception:

pass

self.check_stack_names(sys._getframe(), ['f', 'g'])

def g():

self.check_stack_names(sys._getframe(), ['g'])

yield from f()

self.check_stack_names(sys._getframe(), ['g'])

gen = g()

gen.send(None)

try:

call_method(gen)

except StopIteration:

pass

def test_send_with_yield_from(self):

def call_send(gen):

gen.send(None)

self.check_yield_from_example(call_send)

def test_throw_with_yield_from(self):

def call_throw(gen):

gen.throw(RuntimeError)

self.check_yield_from_example(call_throw)

def test_throw_with_yield_from_custom_generator(self):

class CustomGen:

def __init__(self, test):

self.test = test

def throw(self, *args):

self.test.check_stack_names(sys._getframe(), ['throw', 'g'])

def __iter__(self):

return self

def __next__(self):

return 42

def g(target):

yield from target

gen = g(CustomGen(self))

gen.send(None)

gen.throw(RuntimeError)

class YieldFromTests(unittest.TestCase):

def test_generator_gi_yieldfrom(self):

def a():

self.assertEqual(inspect.getgeneratorstate(gen_b), inspect.GEN_RUNNING)

self.assertIsNone(gen_b.gi_yieldfrom)

yield

self.assertEqual(inspect.getgeneratorstate(gen_b), inspect.GEN_RUNNING)

self.assertIsNone(gen_b.gi_yieldfrom)

def b():

self.assertIsNone(gen_b.gi_yieldfrom)

yield from a()

self.assertIsNone(gen_b.gi_yieldfrom)

yield

self.assertIsNone(gen_b.gi_yieldfrom)

gen_b = b()

self.assertEqual(inspect.getgeneratorstate(gen_b), inspect.GEN_CREATED)

self.assertIsNone(gen_b.gi_yieldfrom)

gen_b.send(None)

self.assertEqual(inspect.getgeneratorstate(gen_b), inspect.GEN_SUSPENDED)

self.assertEqual(gen_b.gi_yieldfrom.gi_code.co_name, 'a')

gen_b.send(None)

self.assertEqual(inspect.getgeneratorstate(gen_b), inspect.GEN_SUSPENDED)

self.assertIsNone(gen_b.gi_yieldfrom)

[] = gen_b # Exhaust generator