import contextlib

import copy

import inspect

import pickle

import sys

import types

import traceback

import unittest

import warnings

from test import support

from test.support import import_helper

from test.support import warnings_helper

from test.support.script_helper import assert_python_ok

try:

import _testcapi

except ImportError:

_testcapi = None

class AsyncYieldFrom:

def __init__(self, obj):

self.obj = obj

def __await__(self):

yield from self.obj

class AsyncYield:

def __init__(self, value):

self.value = value

def __await__(self):

yield self.value

async def asynciter(iterable):

"""Convert an iterable to an asynchronous iterator."""

for x in iterable:

yield x

def run_async(coro):

assert coro.__class__ in {types.GeneratorType, types.CoroutineType}

buffer = []

result = None

while True:

try:

buffer.append(coro.send(None))

except StopIteration as ex:

result = ex.args[0] if ex.args else None

break

return buffer, result

def run_async__await__(coro):

assert coro.__class__ is types.CoroutineType

aw = coro.__await__()

buffer = []

result = None

i = 0

while True:

try:

if i % 2:

buffer.append(next(aw))

else:

buffer.append(aw.send(None))

i += 1

except StopIteration as ex:

result = ex.args[0] if ex.args else None

break

return buffer, result

@contextlib.contextmanager

def silence_coro_gc():

with warnings.catch_warnings():

warnings.simplefilter("ignore")

yield

support.gc_collect()

class AsyncBadSyntaxTest(unittest.TestCase):

def test_badsyntax_1(self):

samples = [

"""def foo():

await something()

""",

"""await something()""",

"""async def foo():

yield from []

""",

"""async def foo():

await await fut

""",

"""async def foo(a=await something()):

pass

""",

"""async def foo(a:await something()):

pass

""",

"""async def foo():

def bar():

[i async for i in els]

""",

"""async def foo():

def bar():

[await i for i in els]

""",

"""async def foo():

def bar():

[i for i in els

async for b in els]

""",

"""async def foo():

def bar():

[i for i in els

for c in b

async for b in els]

""",

"""async def foo():

def bar():

[i for i in els

async for b in els

for c in b]

""",

"""async def foo():

def bar():

[[async for i in b] for b in els]

""",

"""async def foo():

def bar():

[i for i in els

for b in await els]

""",

"""async def foo():

def bar():

[i for i in els

for b in els

if await b]

""",

"""async def foo():

def bar():

[i for i in await els]

""",

"""async def foo():

def bar():

[i for i in els if await i]

""",

"""def bar():

[i async for i in els]

""",

"""def bar():

{i: i async for i in els}

""",

"""def bar():

{i async for i in els}

""",

"""def bar():

[await i for i in els]

""",

"""def bar():

[i for i in els

async for b in els]

""",

"""def bar():

[i for i in els

for c in b

async for b in els]

""",

"""def bar():

[i for i in els

async for b in els

for c in b]

""",

"""def bar():

[i for i in els

for b in await els]

""",

"""def bar():

[i for i in els

for b in els

if await b]

""",

"""def bar():

[i for i in await els]

""",

"""def bar():

[i for i in els if await i]

""",

"""def bar():

[[i async for i in a] for a in elts]

""",

"""[[i async for i in a] for a in elts]

""",

"""async def foo():

await

""",

"""async def foo():

def bar(): pass

await = 1

""",

"""async def foo():

def bar(): pass

await = 1

""",

"""async def foo():

def bar(): pass

if 1:

await = 1

""",

"""def foo():

async def bar(): pass

if 1:

await a

""",

"""def foo():

async def bar(): pass

await a

""",

"""def foo():

def baz(): pass

async def bar(): pass

await a

""",

"""def foo():

def baz(): pass

# 456

async def bar(): pass

# 123

await a

""",

"""async def foo():

def baz(): pass

# 456

async def bar(): pass

# 123

await = 2

""",

"""def foo():

def baz(): pass

async def bar(): pass

await a

""",

"""async def foo():

def baz(): pass

async def bar(): pass

await = 2

""",

"""async def foo():

def async(): pass

""",

"""async def foo():

def await(): pass

""",

"""async def foo():

def bar():

await

""",

"""async def foo():

return lambda async: await

""",

"""async def foo():

return lambda a: await

""",

"""await a()""",

"""async def foo(a=await b):

pass

""",

"""async def foo(a:await b):

pass

""",

"""def baz():

async def foo(a=await b):

pass

""",

"""async def foo(async):

pass

""",

"""async def foo():

def bar():

def baz():

async = 1

""",

"""async def foo():

def bar():

def baz():

pass

async = 1

""",

"""def foo():

async def bar():

async def baz():

pass

def baz():

42

async = 1

""",

"""async def foo():

def bar():

def baz():

pass\nawait foo()

""",

"""def foo():

def bar():

async def baz():

pass\nawait foo()

""",

"""async def foo(await):

pass

""",

"""def foo():

async def bar(): pass

await a

""",

"""def foo():

async def bar():

pass\nawait a

""",

"""def foo():

async for i in arange(2):

pass

""",

"""def foo():

async with resource:

pass

""",

"""async with resource:

pass

""",

"""async for i in arange(2):

pass

""",

]

for code in samples:

with self.subTest(code=code), self.assertRaises(SyntaxError):

compile(code, "<test>", "exec")

def test_badsyntax_2(self):

samples = [

"""def foo():

await = 1

""",

"""class Bar:

def async(): pass

""",

"""class Bar:

async = 1

""",

"""class async:

pass

""",

"""class await:

pass

""",

"""import math as await""",

"""def async():

pass""",

"""def foo(*, await=1):

pass"""

"""async = 1""",

"""print(await=1)"""

]

for code in samples:

with self.subTest(code=code), self.assertRaises(SyntaxError):

compile(code, "<test>", "exec")

def test_badsyntax_3(self):

with self.assertRaises(SyntaxError):

compile("async = 1", "<test>", "exec")

def test_badsyntax_4(self):

samples = [

'''def foo(await):

async def foo(): pass

async def foo():

pass

return await + 1

''',

'''def foo(await):

async def foo(): pass

async def foo(): pass

return await + 1

''',

'''def foo(await):

async def foo(): pass

async def foo(): pass

return await + 1

''',

'''def foo(await):

"""spam"""

async def foo(): \

pass

# 123

async def foo(): pass

# 456

return await + 1

''',

'''def foo(await):

def foo(): pass

def foo(): pass

async def bar(): return await_

await_ = await

try:

bar().send(None)

except StopIteration as ex:

return ex.args[0] + 1

'''

]

for code in samples:

with self.subTest(code=code), self.assertRaises(SyntaxError):

compile(code, "<test>", "exec")

class TokenizerRegrTest(unittest.TestCase):

def test_oneline_defs(self):

buf = []

for i in range(500):

buf.append('def i{i}(): return {i}'.format(i=i))

buf = '\n'.join(buf)

# Test that 500 consequent, one-line defs is OK

ns = {}

exec(buf, ns, ns)

self.assertEqual(ns['i499'](), 499)

# Test that 500 consequent, one-line defs *and*

# one 'async def' following them is OK

buf += '\nasync def foo():\n return'

ns = {}

exec(buf, ns, ns)

self.assertEqual(ns['i499'](), 499)

self.assertTrue(inspect.iscoroutinefunction(ns['foo']))

class CoroutineTest(unittest.TestCase):

def test_gen_1(self):

def gen(): yield

self.assertNotHasAttr(gen, '__await__')

def test_func_1(self):

async def foo():

return 10

f = foo()

self.assertIsInstance(f, types.CoroutineType)

self.assertTrue(bool(foo.__code__.co_flags & inspect.CO_COROUTINE))

self.assertFalse(bool(foo.__code__.co_flags & inspect.CO_GENERATOR))

self.assertTrue(bool(f.cr_code.co_flags & inspect.CO_COROUTINE))

self.assertFalse(bool(f.cr_code.co_flags & inspect.CO_GENERATOR))

self.assertEqual(run_async(f), ([], 10))

self.assertEqual(run_async__await__(foo()), ([], 10))

def bar(): pass

self.assertFalse(bool(bar.__code__.co_flags & inspect.CO_COROUTINE))

def test_func_2(self):

async def foo():

raise StopIteration

with self.assertRaisesRegex(

RuntimeError, "coroutine raised StopIteration"):

run_async(foo())

def test_func_3(self):

async def foo():

raise StopIteration

coro = foo()

self.assertRegex(repr(coro), '^<coroutine object.* at 0x.*>$')

coro.close()

def test_func_4(self):

async def foo():

raise StopIteration

coro = foo()

check = lambda: self.assertRaisesRegex(

TypeError, "'coroutine' object is not iterable")

with check():

list(coro)

with check():

tuple(coro)

with check():

sum(coro)

with check():

iter(coro)

with check():

for i in coro:

pass

with check():

[i for i in coro]

coro.close()

def test_func_5(self):

@types.coroutine

def bar():

yield 1

async def foo():

await bar()

check = lambda: self.assertRaisesRegex(

TypeError, "'coroutine' object is not iterable")

coro = foo()

with check():

for el in coro:

pass

coro.close()

# the following should pass without an error

for el in bar():

self.assertEqual(el, 1)

self.assertEqual([el for el in bar()], [1])

self.assertEqual(tuple(bar()), (1,))

self.assertEqual(next(iter(bar())), 1)

def test_func_6(self):

@types.coroutine

def bar():

yield 1

yield 2

async def foo():

await bar()

f = foo()

self.assertEqual(f.send(None), 1)

self.assertEqual(f.send(None), 2)

with self.assertRaises(StopIteration):

f.send(None)

def test_func_7(self):

async def bar():

return 10

coro = bar()

def foo():

yield from coro

with self.assertRaisesRegex(

TypeError,

"cannot 'yield from' a coroutine object in "

"a non-coroutine generator"):

list(foo())

coro.close()

def test_func_8(self):

@types.coroutine

def bar():

return (yield from coro)

async def foo():

return 'spam'

coro = foo()

self.assertEqual(run_async(bar()), ([], 'spam'))

coro.close()

def test_func_9(self):

async def foo():

pass

with self.assertWarnsRegex(

RuntimeWarning,

r"coroutine '.*test_func_9.*foo' was never awaited"):

foo()

support.gc_collect()

with self.assertWarnsRegex(

RuntimeWarning,

r"coroutine '.*test_func_9.*foo' was never awaited"):

with self.assertRaises(TypeError):

# See bpo-32703.

for _ in foo():

pass

support.gc_collect()

def test_func_10(self):

N = 0

@types.coroutine

def gen():

nonlocal N

try:

a = yield

yield (a ** 2)

except ZeroDivisionError:

N += 100

raise

finally:

N += 1

async def foo():

await gen()

coro = foo()

aw = coro.__await__()

self.assertIs(aw, iter(aw))

next(aw)

self.assertEqual(aw.send(10), 100)

self.assertEqual(N, 0)

aw.close()

self.assertEqual(N, 1)

coro = foo()

aw = coro.__await__()

next(aw)

with self.assertRaises(ZeroDivisionError):

aw.throw(ZeroDivisionError())

self.assertEqual(N, 102)

coro = foo()

aw = coro.__await__()

next(aw)

with self.assertRaises(ZeroDivisionError):

with self.assertWarns(DeprecationWarning):

aw.throw(ZeroDivisionError, ZeroDivisionError(), None)

def test_func_11(self):

async def func(): pass

coro = func()

# Test that PyCoro_Type and _PyCoroWrapper_Type types were properly

# initialized

self.assertIn('__await__', dir(coro))

self.assertIn('__iter__', dir(coro.__await__()))

self.assertIn('coroutine_wrapper', repr(coro.__await__()))

coro.close() # avoid RuntimeWarning

def test_func_12(self):

async def g():

me.send(None)

await foo

me = g()

with self.assertRaisesRegex(ValueError,

"coroutine already executing"):

me.send(None)

def test_func_13(self):

async def g():

pass

coro = g()

with self.assertRaisesRegex(

TypeError,

"can't send non-None value to a just-started coroutine"):

coro.send('spam')

coro.close()

def test_func_14(self):

@types.coroutine

def gen():

yield

async def coro():

try:

await gen()

except GeneratorExit:

await gen()

c = coro()

c.send(None)

with self.assertRaisesRegex(RuntimeError,

"coroutine ignored GeneratorExit"):

c.close()

def test_func_15(self):

# See http://bugs.python.org/issue25887 for details

async def spammer():

return 'spam'

async def reader(coro):

return await coro

spammer_coro = spammer()

with self.assertRaisesRegex(StopIteration, 'spam'):

reader(spammer_coro).send(None)

with self.assertRaisesRegex(RuntimeError,

'cannot reuse already awaited coroutine'):

reader(spammer_coro).send(None)

def test_func_16(self):

# See http://bugs.python.org/issue25887 for details

@types.coroutine

def nop():

yield

async def send():

await nop()

return 'spam'

async def read(coro):

await nop()

return await coro

spammer = send()

reader = read(spammer)

reader.send(None)

reader.send(None)

with self.assertRaisesRegex(Exception, 'ham'):

reader.throw(Exception('ham'))

reader = read(spammer)

reader.send(None)

with self.assertRaisesRegex(RuntimeError,

'cannot reuse already awaited coroutine'):

reader.send(None)

with self.assertRaisesRegex(RuntimeError,

'cannot reuse already awaited coroutine'):

reader.throw(Exception('wat'))

def test_func_17(self):

# See http://bugs.python.org/issue25887 for details

async def coroutine():

return 'spam'

coro = coroutine()

with self.assertRaisesRegex(StopIteration, 'spam'):

coro.send(None)

with self.assertRaisesRegex(RuntimeError,

'cannot reuse already awaited coroutine'):

coro.send(None)

with self.assertRaisesRegex(RuntimeError,

'cannot reuse already awaited coroutine'):

coro.throw(Exception('wat'))

# Closing a coroutine shouldn't raise any exception even if it's

# already closed/exhausted (similar to generators)

coro.close()

coro.close()

def test_func_18(self):

# See http://bugs.python.org/issue25887 for details

async def coroutine():

return 'spam'

coro = coroutine()

await_iter = coro.__await__()

it = iter(await_iter)

with self.assertRaisesRegex(StopIteration, 'spam'):

it.send(None)

with self.assertRaisesRegex(RuntimeError,

'cannot reuse already awaited coroutine'):

it.send(None)

with self.assertRaisesRegex(RuntimeError,

'cannot reuse already awaited coroutine'):

# Although the iterator protocol requires iterators to

# raise another StopIteration here, we don't want to do

# that. In this particular case, the iterator will raise

# a RuntimeError, so that 'yield from' and 'await'

# expressions will trigger the error, instead of silently

# ignoring the call.

next(it)

with self.assertRaisesRegex(RuntimeError,

'cannot reuse already awaited coroutine'):

it.throw(Exception('wat'))

with self.assertRaisesRegex(RuntimeError,

'cannot reuse already awaited coroutine'):

it.throw(Exception('wat'))

# Closing a coroutine shouldn't raise any exception even if it's

# already closed/exhausted (similar to generators)

it.close()

it.close()

def test_func_19(self):

CHK = 0

@types.coroutine

def foo():

nonlocal CHK

yield

try:

yield

except GeneratorExit:

CHK += 1

async def coroutine():

await foo()

coro = coroutine()

coro.send(None)

coro.send(None)

self.assertEqual(CHK, 0)

coro.close()

self.assertEqual(CHK, 1)

for _ in range(3):

# Closing a coroutine shouldn't raise any exception even if it's

# already closed/exhausted (similar to generators)

coro.close()

self.assertEqual(CHK, 1)

def test_coro_wrapper_send_tuple(self):

async def foo():

return (10,)

result = run_async__await__(foo())

self.assertEqual(result, ([], (10,)))

def test_coro_wrapper_send_stop_iterator(self):

async def foo():

return StopIteration(10)

result = run_async__await__(foo())

self.assertIsInstance(result[1], StopIteration)

self.assertEqual(result[1].value, 10)

def test_cr_await(self):

@types.coroutine

def a():

self.assertEqual(inspect.getcoroutinestate(coro_b), inspect.CORO_RUNNING)

self.assertIsNone(coro_b.cr_await)

yield

self.assertEqual(inspect.getcoroutinestate(coro_b), inspect.CORO_RUNNING)

self.assertIsNone(coro_b.cr_await)

async def c():

await a()

async def b():

self.assertIsNone(coro_b.cr_await)

await c()

self.assertIsNone(coro_b.cr_await)

coro_b = b()

self.assertEqual(inspect.getcoroutinestate(coro_b), inspect.CORO_CREATED)

self.assertIsNone(coro_b.cr_await)

coro_b.send(None)

self.assertEqual(inspect.getcoroutinestate(coro_b), inspect.CORO_SUSPENDED)

self.assertEqual(coro_b.cr_await.cr_await.gi_code.co_name, 'a')

with self.assertRaises(StopIteration):

coro_b.send(None) # complete coroutine

self.assertEqual(inspect.getcoroutinestate(coro_b), inspect.CORO_CLOSED)

self.assertIsNone(coro_b.cr_await)

def test_corotype_1(self):

ct = types.CoroutineType

if not support.MISSING_C_DOCSTRINGS:

self.assertIn('into coroutine', ct.send.__doc__)

self.assertIn('inside coroutine', ct.close.__doc__)

self.assertIn('in coroutine', ct.throw.__doc__)

self.assertIn('of the coroutine', ct.__dict__['__name__'].__doc__)

self.assertIn('of the coroutine', ct.__dict__['__qualname__'].__doc__)

self.assertEqual(ct.__name__, 'coroutine')

async def f(): pass

c = f()

self.assertIn('coroutine object', repr(c))

c.close()

def test_await_1(self):

async def foo():

await 1

with self.assertRaisesRegex(TypeError, "'int' object can.t be awaited"):

run_async(foo())

def test_await_2(self):

async def foo():

await []

with self.assertRaisesRegex(TypeError, "'list' object can.t be awaited"):

run_async(foo())

def test_await_3(self):

async def foo():

await AsyncYieldFrom([1, 2, 3])

self.assertEqual(run_async(foo()), ([1, 2, 3], None))

self.assertEqual(run_async__await__(foo()), ([1, 2, 3], None))

def test_await_4(self):

async def bar():

return 42

async def foo():

return await bar()

self.assertEqual(run_async(foo()), ([], 42))