import pytest
|
|
import types
|
import sys
|
import collections.abc
|
from functools import wraps
|
import gc
|
|
from .conftest import mock_sleep
|
from .. import (
|
async_generator,
|
yield_,
|
yield_from_,
|
isasyncgen,
|
isasyncgenfunction,
|
get_asyncgen_hooks,
|
set_asyncgen_hooks,
|
)
|
|
|
# like list(it) but works on async iterators
|
async def collect(ait):
|
items = []
|
async for value in ait:
|
items.append(value)
|
return items
|
|
|
################################################################
|
#
|
# Basic test
|
#
|
################################################################
|
|
|
@async_generator
|
async def async_range(count):
|
for i in range(count):
|
print("Calling yield_({})".format(i))
|
await yield_(i)
|
|
|
@async_generator
|
async def double(ait):
|
async for value in ait:
|
await yield_(value * 2)
|
await mock_sleep()
|
|
|
class HasAsyncGenMethod:
|
def __init__(self, factor):
|
self._factor = factor
|
|
@async_generator
|
async def async_multiplied(self, ait):
|
async for value in ait:
|
await yield_(value * self._factor)
|
|
|
async def test_async_generator():
|
assert await collect(async_range(10)) == list(range(10))
|
assert (await collect(double(async_range(5))) == [0, 2, 4, 6, 8])
|
|
tripler = HasAsyncGenMethod(3)
|
assert (
|
await
|
collect(tripler.async_multiplied(async_range(5))) == [0, 3, 6, 9, 12]
|
)
|
|
|
@async_generator
|
async def agen_yield_no_arg():
|
await yield_()
|
|
|
async def test_yield_no_arg():
|
assert await collect(agen_yield_no_arg()) == [None]
|
|
|
################################################################
|
#
|
# async_generators return value
|
#
|
################################################################
|
|
|
@async_generator
|
async def async_gen_with_non_None_return():
|
await yield_(1)
|
await yield_(2)
|
return "hi"
|
|
|
async def test_bad_return_value():
|
gen = async_gen_with_non_None_return()
|
async for item in gen: # pragma: no branch
|
assert item == 1
|
break
|
async for item in gen: # pragma: no branch
|
assert item == 2
|
break
|
try:
|
await gen.__anext__()
|
except StopAsyncIteration as e:
|
assert e.args[0] == "hi"
|
|
|
################################################################
|
#
|
# Exhausitve tests of the different ways to re-enter a coroutine.
|
#
|
# It used to be that re-entering via send/__next__ would work, but throw()
|
# immediately followed by an await yield_(...) wouldn't work, and the
|
# YieldWrapper object would propagate back out to the coroutine runner.
|
#
|
# Before I fixed this, the 'assert value is None' check below would fail
|
# (because of the YieldWrapper leaking out), and if you removed that
|
# assertion, then the code would appear to run successfully but the final list
|
# would just be [1, 3] instead of [1, 2, 3].
|
#
|
################################################################
|
|
|
class MyTestError(Exception):
|
pass
|
|
|
# This unconditionally raises a MyTestError exception, so from the outside
|
# it's equivalent to a simple 'raise MyTestError`. But, for this test to check
|
# the thing we want it to check, the point is that the exception must be
|
# thrown in from the coroutine runner -- this simulates something like an
|
# 'await sock.recv(...) -> TimeoutError'.
|
@types.coroutine
|
def hit_me():
|
yield "hit me"
|
|
|
@types.coroutine
|
def number_me():
|
assert (yield "number me") == 1
|
|
|
@types.coroutine
|
def next_me():
|
assert (yield "next me") is None
|
|
|
@async_generator
|
async def yield_after_different_entries():
|
await yield_(1)
|
try:
|
await hit_me()
|
except MyTestError:
|
await yield_(2)
|
await number_me()
|
await yield_(3)
|
await next_me()
|
await yield_(4)
|
|
|
def hostile_coroutine_runner(coro):
|
coro_iter = coro.__await__()
|
value = None
|
while True:
|
try:
|
if value == "hit me":
|
value = coro_iter.throw(MyTestError())
|
elif value == "number me":
|
value = coro_iter.send(1)
|
else:
|
assert value in (None, "next me")
|
value = coro_iter.__next__()
|
except StopIteration as exc:
|
return exc.value
|
|
|
def test_yield_different_entries():
|
coro = collect(yield_after_different_entries())
|
yielded = hostile_coroutine_runner(coro)
|
assert yielded == [1, 2, 3, 4]
|
|
|
async def test_reentrance_forbidden():
|
@async_generator
|
async def recurse():
|
async for obj in agen: # pragma: no branch
|
await yield_(obj) # pragma: no cover
|
|
agen = recurse()
|
with pytest.raises(ValueError):
|
async for _ in agen: # pragma: no branch
|
pass # pragma: no cover
|
|
|
async def test_reentrance_forbidden_simultaneous_asends():
|
@async_generator
|
async def f():
|
await mock_sleep()
|
|
ag = f()
|
sender1 = ag.asend(None)
|
sender2 = ag.asend(None)
|
assert sender1.send(None) == "mock_sleep"
|
with pytest.raises(ValueError):
|
sender2.send(None)
|
with pytest.raises(StopAsyncIteration):
|
sender1.send(None)
|
await ag.aclose()
|
|
|
# https://bugs.python.org/issue32526
|
async def test_reentrance_forbidden_while_suspended_in_coroutine_runner():
|
@async_generator
|
async def f():
|
await mock_sleep()
|
await yield_("final yield")
|
|
ag = f()
|
asend_coro = ag.asend(None)
|
fut = asend_coro.send(None)
|
assert fut == "mock_sleep"
|
# Now the async generator's frame is not executing, but a call to asend()
|
# *is* executing. Make sure that in this case, ag_running is True, and we
|
# can't start up another call to asend().
|
assert ag.ag_running
|
with pytest.raises(ValueError):
|
await ag.asend(None)
|
# Clean up
|
with pytest.raises(StopIteration):
|
asend_coro.send(None)
|
with pytest.raises(StopAsyncIteration):
|
ag.asend(None).send(None)
|
|
|
################################################################
|
#
|
# asend
|
#
|
################################################################
|
|
|
@async_generator
|
async def asend_me():
|
assert (await yield_(1)) == 2
|
assert (await yield_(3)) == 4
|
|
|
async def test_asend():
|
aiter = asend_me()
|
assert (await aiter.__anext__()) == 1
|
assert (await aiter.asend(2)) == 3
|
with pytest.raises(StopAsyncIteration):
|
await aiter.asend(4)
|
|
|
################################################################
|
#
|
# athrow
|
#
|
################################################################
|
|
|
@async_generator
|
async def athrow_me():
|
with pytest.raises(KeyError):
|
await yield_(1)
|
with pytest.raises(ValueError):
|
await yield_(2)
|
await yield_(3)
|
|
|
async def test_athrow():
|
aiter = athrow_me()
|
assert (await aiter.__anext__()) == 1
|
assert (await aiter.athrow(KeyError("oops"))) == 2
|
assert (await aiter.athrow(ValueError("oops"))) == 3
|
with pytest.raises(OSError):
|
await aiter.athrow(OSError("oops"))
|
|
|
################################################################
|
#
|
# aclose
|
#
|
################################################################
|
|
|
@async_generator
|
async def close_me_aiter(track):
|
try:
|
await yield_(1)
|
except GeneratorExit:
|
track[0] = "closed"
|
raise
|
else: # pragma: no cover
|
track[0] = "wtf"
|
|
|
async def test_aclose():
|
track = [None]
|
aiter = close_me_aiter(track)
|
async for obj in aiter: # pragma: no branch
|
assert obj == 1
|
break
|
assert track[0] is None
|
await aiter.aclose()
|
assert track[0] == "closed"
|
|
|
async def test_aclose_on_unstarted_generator():
|
aiter = close_me_aiter([None])
|
await aiter.aclose()
|
async for obj in aiter:
|
assert False # pragma: no cover
|
|
|
async def test_aclose_on_finished_generator():
|
aiter = async_range(3)
|
async for obj in aiter:
|
pass # pragma: no cover
|
await aiter.aclose()
|
|
|
@async_generator
|
async def sync_yield_during_aclose():
|
try:
|
await yield_(1)
|
finally:
|
await mock_sleep()
|
|
|
@async_generator
|
async def async_yield_during_aclose():
|
try:
|
await yield_(1)
|
finally:
|
await yield_(2)
|
|
|
async def test_aclose_yielding():
|
aiter = sync_yield_during_aclose()
|
assert (await aiter.__anext__()) == 1
|
# Doesn't raise:
|
await aiter.aclose()
|
|
aiter = async_yield_during_aclose()
|
assert (await aiter.__anext__()) == 1
|
with pytest.raises(RuntimeError):
|
await aiter.aclose()
|
|
|
################################################################
|
#
|
# yield from
|
#
|
################################################################
|
|
|
@async_generator
|
async def async_range_twice(count):
|
await yield_from_(async_range(count))
|
await yield_(None)
|
await yield_from_(async_range(count))
|
|
|
if sys.version_info >= (3, 6):
|
exec(
|
"""
|
async def native_async_range(count):
|
for i in range(count):
|
yield i
|
|
# XX uncomment if/when we re-enable the ctypes hacks:
|
# async def native_async_range_twice(count):
|
# # make sure yield_from_ works inside a native async generator
|
# await yield_from_(async_range(count))
|
# yield None
|
# # make sure we can yield_from_ a native async generator
|
# await yield_from_(native_async_range(count))
|
"""
|
)
|
|
|
async def test_async_yield_from_():
|
assert await collect(async_range_twice(3)) == [
|
0,
|
1,
|
2,
|
None,
|
0,
|
1,
|
2,
|
]
|
|
if sys.version_info >= (3, 6):
|
# Make sure we can yield_from_ a native generator
|
@async_generator
|
async def yield_from_native():
|
await yield_from_(native_async_range(3))
|
|
assert await collect(yield_from_native()) == [0, 1, 2]
|
|
# XX uncomment if/when we re-enable the ctypes hacks:
|
# if sys.version_info >= (3, 6):
|
# assert await collect(native_async_range_twice(3)) == [
|
# 0, 1, 2, None, 0, 1, 2,
|
# ]
|
|
|
@async_generator
|
async def doubles_sends(value):
|
while True:
|
value = await yield_(2 * value)
|
|
|
@async_generator
|
async def wraps_doubles_sends(value):
|
await yield_from_(doubles_sends(value))
|
|
|
async def test_async_yield_from_asend():
|
gen = wraps_doubles_sends(10)
|
await gen.__anext__() == 20
|
assert (await gen.asend(2)) == 4
|
assert (await gen.asend(5)) == 10
|
assert (await gen.asend(0)) == 0
|
await gen.aclose()
|
|
|
async def test_async_yield_from_athrow():
|
gen = async_range_twice(2)
|
assert (await gen.__anext__()) == 0
|
with pytest.raises(ValueError):
|
await gen.athrow(ValueError)
|
|
|
@async_generator
|
async def returns_1():
|
await yield_(0)
|
return 1
|
|
|
@async_generator
|
async def yields_from_returns_1():
|
await yield_(await yield_from_(returns_1()))
|
|
|
async def test_async_yield_from_return_value():
|
assert await collect(yields_from_returns_1()) == [0, 1]
|
|
|
# Special cases to get coverage
|
async def test_yield_from_empty():
|
@async_generator
|
async def empty():
|
return "done"
|
|
@async_generator
|
async def yield_from_empty():
|
assert (await yield_from_(empty())) == "done"
|
|
assert await collect(yield_from_empty()) == []
|
|
|
async def test_yield_from_non_generator():
|
class Countdown:
|
def __init__(self, count):
|
self.count = count
|
self.closed = False
|
|
if sys.version_info < (3, 5, 2):
|
|
async def __aiter__(self):
|
return self
|
else:
|
|
def __aiter__(self):
|
return self
|
|
async def __anext__(self):
|
self.count -= 1
|
if self.count < 0:
|
raise StopAsyncIteration("boom")
|
return self.count
|
|
async def aclose(self):
|
self.closed = True
|
|
@async_generator
|
async def yield_from_countdown(count, happenings):
|
try:
|
c = Countdown(count)
|
assert (await yield_from_(c)) == "boom"
|
except BaseException as e:
|
if c.closed:
|
happenings.append("countdown closed")
|
happenings.append("raise")
|
return e
|
|
h = []
|
assert await collect(yield_from_countdown(3, h)) == [2, 1, 0]
|
assert h == []
|
|
# Throwing into a yield_from_(object with no athrow) just raises the
|
# exception in the generator.
|
h = []
|
agen = yield_from_countdown(3, h)
|
assert await agen.__anext__() == 2
|
exc = ValueError("x")
|
try:
|
await agen.athrow(exc)
|
except StopAsyncIteration as e:
|
assert e.args[0] is exc
|
assert h == ["raise"]
|
|
# Calling aclose on the generator calls aclose on the iterator
|
h = []
|
agen = yield_from_countdown(3, h)
|
assert await agen.__anext__() == 2
|
await agen.aclose()
|
assert h == ["countdown closed", "raise"]
|
|
# Throwing GeneratorExit into the generator calls *aclose* on the iterator
|
# (!)
|
h = []
|
agen = yield_from_countdown(3, h)
|
assert await agen.__anext__() == 2
|
exc = GeneratorExit()
|
with pytest.raises(StopAsyncIteration):
|
await agen.athrow(exc)
|
assert h == ["countdown closed", "raise"]
|
|
|
async def test_yield_from_non_generator_with_no_aclose():
|
class Countdown:
|
def __init__(self, count):
|
self.count = count
|
self.closed = False
|
|
if sys.version_info < (3, 5, 2):
|
|
async def __aiter__(self):
|
return self
|
else:
|
|
def __aiter__(self):
|
return self
|
|
async def __anext__(self):
|
self.count -= 1
|
if self.count < 0:
|
raise StopAsyncIteration("boom")
|
return self.count
|
|
@async_generator
|
async def yield_from_countdown(count):
|
return await yield_from_(Countdown(count))
|
|
assert await collect(yield_from_countdown(3)) == [2, 1, 0]
|
|
agen = yield_from_countdown(3)
|
assert await agen.__anext__() == 2
|
assert await agen.__anext__() == 1
|
# It's OK that Countdown has no aclose
|
await agen.aclose()
|
|
|
async def test_yield_from_with_old_style_aiter():
|
# old-style 'async def __aiter__' should still work even on newer pythons
|
class Countdown:
|
def __init__(self, count):
|
self.count = count
|
self.closed = False
|
|
# This is wrong, that's the point
|
async def __aiter__(self):
|
return self
|
|
async def __anext__(self):
|
self.count -= 1
|
if self.count < 0:
|
raise StopAsyncIteration("boom")
|
return self.count
|
|
@async_generator
|
async def yield_from_countdown(count):
|
return await yield_from_(Countdown(count))
|
|
assert await collect(yield_from_countdown(3)) == [2, 1, 0]
|
|
|
async def test_yield_from_athrow_raises_StopAsyncIteration():
|
@async_generator
|
async def catch():
|
try:
|
while True:
|
await yield_("hi")
|
except Exception as exc:
|
return ("bye", exc)
|
|
@async_generator
|
async def yield_from_catch():
|
return await yield_from_(catch())
|
|
agen = yield_from_catch()
|
assert await agen.__anext__() == "hi"
|
assert await agen.__anext__() == "hi"
|
thrown = ValueError("oops")
|
try:
|
print(await agen.athrow(thrown))
|
except StopAsyncIteration as caught:
|
assert caught.args == (("bye", thrown),)
|
else:
|
raise AssertionError # pragma: no cover
|
|
|
################################################################
|
# __del__
|
################################################################
|
|
|
async def test___del__(capfd):
|
completions = 0
|
|
@async_generator
|
async def awaits_when_unwinding():
|
await yield_(0)
|
try:
|
await yield_(1)
|
finally:
|
await mock_sleep()
|
try:
|
await yield_(2)
|
finally:
|
nonlocal completions
|
completions += 1
|
|
gen = awaits_when_unwinding()
|
# Hasn't started yet, so no problem
|
gen.__del__()
|
|
gen = awaits_when_unwinding()
|
assert await collect(gen) == [0, 1, 2]
|
# Exhausted, so no problem
|
gen.__del__()
|
|
for stop_after_turn in (1, 2, 3):
|
gen = awaits_when_unwinding()
|
for turn in range(stop_after_turn):
|
assert await gen.__anext__() == turn
|
await gen.aclose()
|
# Closed, so no problem
|
gen.__del__()
|
|
for stop_after_turn in (1, 2, 3):
|
gen = awaits_when_unwinding()
|
for turn in range(stop_after_turn):
|
assert await gen.__anext__() == turn
|
|
if stop_after_turn == 2:
|
# Stopped in the middle of a try/finally that awaits in the finally,
|
# so __del__ can't cleanup.
|
with pytest.raises(RuntimeError) as info:
|
gen.__del__()
|
assert "awaited during finalization; install a finalization hook" in str(
|
info.value
|
)
|
else:
|
# Can clean up without awaiting, so __del__ is fine
|
gen.__del__()
|
|
assert completions == 3
|
|
@async_generator
|
async def yields_when_unwinding():
|
try:
|
await yield_(1)
|
finally:
|
await yield_(2)
|
|
gen = yields_when_unwinding()
|
assert await gen.__anext__() == 1
|
with pytest.raises(RuntimeError) as info:
|
gen.__del__()
|
|
|
################################################################
|
# introspection
|
################################################################
|
|
|
def test_isasyncgen():
|
assert not isasyncgen(async_range)
|
assert isasyncgen(async_range(10))
|
|
if sys.version_info >= (3, 6):
|
assert not isasyncgen(native_async_range)
|
assert isasyncgen(native_async_range(10))
|
|
|
def test_isasyncgenfunction():
|
assert isasyncgenfunction(async_range)
|
assert not isasyncgenfunction(list)
|
assert not isasyncgenfunction(async_range(10))
|
|
if sys.version_info >= (3, 6):
|
assert isasyncgenfunction(native_async_range)
|
assert not isasyncgenfunction(native_async_range(10))
|
|
|
# Very subtle bug: functools.wraps copies across the entire contents of the
|
# wrapped function's __dict__. We used to use a simple _is_async_gen=True
|
# attribute to mark async generators. But if we do that, then simple wrappers
|
# like async_range_wrapper *do* return True for isasyncgenfunction. But that's
|
# not how inspect.isasyncgenfunction works, and it also caused problems for
|
# sphinxcontrib-trio, because given a function like:
|
#
|
# @acontextmanager
|
# @async_generator
|
# async def async_cm():
|
# ...
|
#
|
# then we end up with async_cm introspecting as both an async context manager
|
# and an async generator, and it doesn't know who to believe. With the
|
# correct, inspect.isasyncgenfunction-compliant behavior, we have async_cm
|
# introspecting as an async context manager, and async_cm.__wrapped__
|
# introspecting as an async generator.
|
def test_isasyncgenfunction_is_not_inherited_by_wrappers():
|
@wraps(async_range)
|
def async_range_wrapper(*args, **kwargs): # pragma: no cover
|
return async_range(*args, **kwargs)
|
|
assert not isasyncgenfunction(async_range_wrapper)
|
assert isasyncgenfunction(async_range_wrapper.__wrapped__)
|
|
|
def test_collections_abc_AsyncGenerator():
|
if hasattr(collections.abc, "AsyncGenerator"):
|
assert isinstance(async_range(10), collections.abc.AsyncGenerator)
|
|
|
async def test_ag_attributes():
|
@async_generator
|
async def f():
|
x = 1
|
await yield_()
|
|
agen = f()
|
assert agen.ag_code.co_name == "f"
|
async for _ in agen: # pragma: no branch
|
assert agen.ag_frame.f_locals["x"] == 1
|
break
|
|
|
################################################################
|
# Finicky tests to check that the overly clever ctype stuff has plausible
|
# refcounting
|
|
from .. import _impl
|
|
|
@pytest.mark.skipif(not hasattr(sys, "getrefcount"), reason="CPython only")
|
def test_refcnt():
|
x = object()
|
print(sys.getrefcount(x))
|
print(sys.getrefcount(x))
|
print(sys.getrefcount(x))
|
print(sys.getrefcount(x))
|
base_count = sys.getrefcount(x)
|
l = [_impl._wrap(x) for _ in range(100)]
|
print(sys.getrefcount(x))
|
print(sys.getrefcount(x))
|
print(sys.getrefcount(x))
|
assert sys.getrefcount(x) >= base_count + 100
|
l2 = [_impl._unwrap(box) for box in l]
|
assert sys.getrefcount(x) >= base_count + 200
|
print(sys.getrefcount(x))
|
print(sys.getrefcount(x))
|
print(sys.getrefcount(x))
|
print(sys.getrefcount(x))
|
del l
|
print(sys.getrefcount(x))
|
print(sys.getrefcount(x))
|
print(sys.getrefcount(x))
|
del l2
|
print(sys.getrefcount(x))
|
print(sys.getrefcount(x))
|
print(sys.getrefcount(x))
|
assert sys.getrefcount(x) == base_count
|
print(sys.getrefcount(x))
|
|
|
################################################################
|
#
|
# Edge cases
|
#
|
################################################################
|
|
|
# PEP 479: StopIteration or StopAsyncIteration exiting from inside an async
|
# generator should produce a RuntimeError with the __cause__ set to the
|
# original exception. Note that contextlib.asynccontextmanager depends on this
|
# behavior.
|
@async_generator
|
async def lets_exception_out():
|
await yield_()
|
|
|
async def test_throw_StopIteration_or_StopAsyncIteration():
|
for cls in [StopIteration, StopAsyncIteration]:
|
agen = lets_exception_out()
|
await agen.asend(None)
|
exc = cls()
|
with pytest.raises(RuntimeError) as excinfo:
|
await agen.athrow(exc)
|
assert excinfo.type is RuntimeError
|
assert excinfo.value.__cause__ is exc
|
|
|
# No "coroutine was never awaited" warnings for async generators that are not
|
# iterated
|
async def test_no_spurious_unawaited_coroutine_warning(recwarn):
|
agen = async_range(10)
|
del agen
|
|
# Run collection a few times to make sure any
|
# loops/resurrection/etc. stuff gets fully handled (necessary on pypy)
|
for _ in range(4):
|
gc.collect()
|
|
# I've seen DeprecationWarnings here triggered by pytest-asyncio, so let's
|
# filter for RuntimeWarning. But if there are no warnings at all, then
|
# that's OK too, so tell coverage not to worry about it.
|
for msg in recwarn: # pragma: no cover
|
print(msg)
|
assert not issubclass(msg.category, RuntimeWarning)
|
|
|
################################################################
|
#
|
# GC hooks
|
#
|
################################################################
|
|
|
@pytest.fixture
|
def local_asyncgen_hooks():
|
old_hooks = get_asyncgen_hooks()
|
yield
|
set_asyncgen_hooks(*old_hooks)
|
|
|
def test_gc_hooks_interface(local_asyncgen_hooks):
|
def one(agen): # pragma: no cover
|
pass
|
|
def two(agen): # pragma: no cover
|
pass
|
|
set_asyncgen_hooks(None, None)
|
assert get_asyncgen_hooks() == (None, None)
|
set_asyncgen_hooks(finalizer=two)
|
assert get_asyncgen_hooks() == (None, two)
|
set_asyncgen_hooks(firstiter=one)
|
assert get_asyncgen_hooks() == (one, two)
|
set_asyncgen_hooks(finalizer=None, firstiter=two)
|
assert get_asyncgen_hooks() == (two, None)
|
set_asyncgen_hooks(None, one)
|
assert get_asyncgen_hooks() == (None, one)
|
tup = (one, two)
|
set_asyncgen_hooks(*tup)
|
assert get_asyncgen_hooks() == tup
|
|
with pytest.raises(TypeError):
|
set_asyncgen_hooks(firstiter=42)
|
|
with pytest.raises(TypeError):
|
set_asyncgen_hooks(finalizer=False)
|
|
def in_thread(results):
|
results.append(get_asyncgen_hooks())
|
set_asyncgen_hooks(two, one)
|
results.append(get_asyncgen_hooks())
|
|
from threading import Thread
|
results = []
|
thread = Thread(target=in_thread, args=(results,))
|
thread.start()
|
thread.join()
|
assert results == [(None, None), (two, one)]
|
assert get_asyncgen_hooks() == (one, two)
|
|
|
async def test_gc_hooks_behavior(local_asyncgen_hooks):
|
events = []
|
to_finalize = []
|
|
def firstiter(agen):
|
events.append("firstiter {}".format(agen.ag_frame.f_locals["ident"]))
|
|
def finalizer(agen):
|
events.append("finalizer {}".format(agen.ag_frame.f_locals["ident"]))
|
to_finalize.append(agen)
|
|
@async_generator
|
async def agen(ident):
|
events.append("yield 1 {}".format(ident))
|
await yield_(1)
|
try:
|
events.append("yield 2 {}".format(ident))
|
await yield_(2)
|
events.append("after yield 2 {}".format(ident))
|
finally:
|
events.append("mock_sleep {}".format(ident))
|
await mock_sleep()
|
try:
|
events.append("yield 3 {}".format(ident))
|
await yield_(3)
|
finally:
|
events.append("unwind 3 {}".format(ident))
|
# this one is included to make sure we _don't_ execute it
|
events.append("done {}".format(ident)) # pragma: no cover
|
|
async def anext_verbosely(iter, ident):
|
events.append("before asend {}".format(ident))
|
sender = iter.asend(None)
|
events.append("before send {}".format(ident))
|
await sender
|
events.append("after asend {}".format(ident))
|
|
# Ensure that firstiter is called immediately on asend(),
|
# before the first turn of the coroutine that asend() returns,
|
# to match the behavior of native generators.
|
# Ensure that the firstiter that gets used is the one in effect
|
# at the time of that first call, rather than at the time of iteration.
|
iterA = agen("A")
|
iterB = agen("B")
|
await anext_verbosely(iterA, "A")
|
set_asyncgen_hooks(firstiter, finalizer)
|
await anext_verbosely(iterB, "B")
|
iterC = agen("C")
|
await anext_verbosely(iterC, "C")
|
|
assert events == [
|
"before asend A", "before send A", "yield 1 A", "after asend A",
|
"before asend B", "firstiter B", "before send B", "yield 1 B",
|
"after asend B", "before asend C", "firstiter C", "before send C",
|
"yield 1 C", "after asend C"
|
]
|
del events[:]
|
|
# Ensure that firstiter is only called once, even if we create
|
# two asend() coroutines before iterating either of them.
|
iterX = agen("X")
|
sender1 = iterX.asend(None)
|
sender2 = iterX.asend(None)
|
events.append("before close")
|
sender1.close()
|
sender2.close()
|
await iterX.aclose()
|
assert events == ["firstiter X", "before close"]
|
del events[:]
|
|
from weakref import ref
|
refA, refB, refC = map(ref, (iterA, iterB, iterC))
|
|
# iterA uses the finalizer that was in effect when it started, i.e. no finalizer
|
await iterA.__anext__()
|
await iterA.__anext__()
|
del iterA
|
# Do multiple GC passes since we're deliberately shielding the
|
# coroutine objects from the first pass due to PyPy issue 2786.
|
for _ in range(4):
|
gc.collect()
|
assert refA() is None
|
assert events == [
|
"yield 2 A", "after yield 2 A", "mock_sleep A", "yield 3 A",
|
"unwind 3 A"
|
]
|
assert not to_finalize
|
del events[:]
|
|
# iterB and iterC do use our finalizer
|
await iterC.__anext__()
|
await iterB.__anext__()
|
await iterC.__anext__()
|
idB, idC = id(iterB), id(iterC)
|
del iterB
|
for _ in range(4):
|
gc.collect()
|
del iterC
|
for _ in range(4):
|
gc.collect()
|
assert events == [
|
"yield 2 C", "yield 2 B", "after yield 2 C", "mock_sleep C",
|
"yield 3 C", "finalizer B", "finalizer C"
|
]
|
del events[:]
|
|
# finalizer invokes aclose() is not called again once the revived reference drops
|
assert list(map(id, to_finalize)) == [idB, idC]
|
events.append("before aclose B")
|
await to_finalize[0].aclose()
|
events.append("before aclose C")
|
await to_finalize[1].aclose()
|
events.append("after aclose both")
|
del to_finalize[:]
|
for _ in range(4):
|
gc.collect()
|
assert refB() is None and refC() is None
|
|
assert events == [
|
"before aclose B", "mock_sleep B", "before aclose C", "unwind 3 C",
|
"after aclose both"
|
]
|
