cpython/Lib/test/test_generators.py at 3.11 · python/cpython

2444 lines (2043 loc) · 64.6 KB

10BC0

import copy

import gc

import pickle

import sys

import doctest

import unittest

import weakref

import inspect

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.assertGreater(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):

# Issue #23192: Test that a lambda returning a generator behaves

# like the equivalent function

f = lambda: (yield 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)

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)

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.assertEqual(sys.exc_info()[0], None)

yield

except Exception as exc:

# exception raised by gen.throw(exc)

self.assertEqual(sys.exc_info()[0], ValueError)

self.assertIsNone(exc.__context__)

yield

# ensure that the exception is not lost

self.assertEqual(sys.exc_info()[0], 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.assertEqual(sys.exc_info(), (None, None, None))

def test_except_next(self):

def gen():

self.assertEqual(sys.exc_info()[0], ValueError)

yield "done"

g = gen()

try:

raise ValueError

except Exception:

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

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

def test_except_gen_except(self):

def gen():

try:

self.assertEqual(sys.exc_info()[0], None)

yield

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

# inherit ValueError in its context

raise TypeError()

except TypeError as exc:

self.assertEqual(sys.exc_info()[0], TypeError)

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

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

self.assertEqual(sys.exc_info()[0], ValueError)

yield

self.assertIsNone(sys.exc_info()[0])

yield "done"

g = gen()

next(g)

try:

raise ValueError

except Exception:

next(g)

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

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

def test_except_throw_exception_context(self):

def gen():

try:

self.assertEqual(sys.exc_info()[0], None)

yield

except ValueError:

# we are called from "except ValueError:"

self.assertEqual(sys.exc_info()[0], ValueError)

raise TypeError()

except Exception as exc:

self.assertEqual(sys.exc_info()[0], TypeError)

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

# we are still called from "except ValueError:"

self.assertEqual(sys.exc_info()[0], ValueError)

yield

self.assertIsNone(sys.exc_info()[0])

yield "done"

g = gen()

next(g)

try:

raise ValueError

except Exception as exc:

g.throw(exc)

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

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

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_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 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_'):

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)

class YieldFromTests(unittest.TestCase):

def test_generator_gi_yieldfrom(self):

def a():

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