Commit c33df193 authored by Damien George's avatar Damien George
Browse files

tests: Add 6 tests for async await/for/with.

parent eacbd7ae
# test basic await expression
# adapted from PEP0492
async def abinary(n):
print(n)
if n <= 0:
return 1
l = await abinary(n - 1)
r = await abinary(n - 1)
return l + 1 + r
o = abinary(4)
try:
while True:
o.send(None)
except StopIteration:
print('finished')
# test await expression
import sys
if sys.implementation.name == 'micropython':
# uPy allows normal generators to be awaitables
coroutine = lambda f: f
else:
import types
coroutine = types.coroutine
@coroutine
def wait(value):
print('wait value:', value)
msg = yield 'message from wait(%u)' % value
print('wait got back:', msg)
return 10
async def f():
x = await wait(1)**2
print('x =', x)
coro = f()
print('return from send:', coro.send(None))
try:
coro.send('message from main')
except StopIteration:
print('got StopIteration')
# test basic async for execution
# example taken from PEP0492
class AsyncIteratorWrapper:
def __init__(self, obj):
print('init')
self._it = iter(obj)
async def __aiter__(self):
print('aiter')
return self
async def __anext__(self):
print('anext')
try:
value = next(self._it)
except StopIteration:
raise StopAsyncIteration
return value
async def coro():
async for letter in AsyncIteratorWrapper('abc'):
print(letter)
o = coro()
try:
o.send(None)
except StopIteration:
print('finished')
# test waiting within "async for" aiter/anext functions
import sys
if sys.implementation.name == 'micropython':
# uPy allows normal generators to be awaitables
coroutine = lambda f: f
else:
import types
coroutine = types.coroutine
@coroutine
def f(x):
print('f start:', x)
yield x + 1
yield x + 2
return x + 3
class ARange:
def __init__(self, high):
print('init')
self.cur = 0
self.high = high
async def __aiter__(self):
print('aiter')
print('f returned:', await f(10))
return self
async def __anext__(self):
print('anext')
print('f returned:', await f(20))
if self.cur < self.high:
val = self.cur
self.cur += 1
return val
else:
raise StopAsyncIteration
async def coro():
async for x in ARange(4):
print('x', x)
o = coro()
try:
while True:
print('coro yielded:', o.send(None))
except StopIteration:
print('finished')
# test simple async with execution
class AContext:
async def __aenter__(self):
print('enter')
async def __aexit__(self, exc_type, exc, tb):
print('exit')
async def f():
async with AContext():
print('body')
o = f()
try:
o.send(None)
except StopIteration:
print('finished')
# test waiting within async with enter/exit functions
import sys
if sys.implementation.name == 'micropython':
# uPy allows normal generators to be awaitables
coroutine = lambda f: f
else:
import types
coroutine = types.coroutine
@coroutine
def f(x):
print('f start:', x)
yield x + 1
yield x + 2
return x + 3
class AContext:
async def __aenter__(self):
print('enter')
print('f returned:', await f(10))
async def __aexit__(self, exc_type, exc, tb):
print('exit')
print('f returned:', await f(20))
async def coro():
async with AContext():
print('body start')
print('body f returned:', await f(30))
print('body end')
o = coro()
try:
while True:
print('coro yielded:', o.send(None))
except StopIteration:
print('finished')
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment