test_pep492.py

"""Tests support for new syntax introduced by PEP 492."""

importsys
importtypes
importunittest

fromunittestimportmock

importasyncio
fromtest.test_asyncioimportutilsastest_utils


deftearDownModule():
asyncio.set_event_loop_policy(None)


# Test that asyncio.iscoroutine() uses collections.abc.Coroutine
classFakeCoro:
defsend(self, value):
pass

defthrow(self, typ, val=None, tb=None):
pass

defclose(self):
pass

def__await__(self):
yield


classBaseTest(test_utils.TestCase):

defsetUp(self):
super().setUp()
self.loop=asyncio.BaseEventLoop()
self.loop._process_events=mock.Mock()
self.loop._selector=mock.Mock()
self.loop._selector.select.return_value= ()
self.set_event_loop(self.loop)


classLockTests(BaseTest):

deftest_context_manager_async_with(self):
primitives= [
asyncio.Lock(),
asyncio.Condition(),
asyncio.Semaphore(),
asyncio.BoundedSemaphore(),
 ]

asyncdeftest(lock):
awaitasyncio.sleep(0.01)
self.assertFalse(lock.locked())
asyncwithlockas_lock:
self.assertIs(_lock, None)
self.assertTrue(lock.locked())
awaitasyncio.sleep(0.01)
self.assertTrue(lock.locked())
self.assertFalse(lock.locked())

forprimitiveinprimitives:
self.loop.run_until_complete(test(primitive))
self.assertFalse(primitive.locked())

deftest_context_manager_with_await(self):
primitives= [
asyncio.Lock(),
asyncio.Condition(),
asyncio.Semaphore(),
asyncio.BoundedSemaphore(),
 ]

asyncdeftest(lock):
awaitasyncio.sleep(0.01)
self.assertFalse(lock.locked())
withself.assertRaisesRegex(
TypeError,
"can't be used in 'await' expression"
 ):
withawaitlock:
pass

forprimitiveinprimitives:
self.loop.run_until_complete(test(primitive))
self.assertFalse(primitive.locked())


classStreamReaderTests(BaseTest):

deftest_readline(self):
DATA=b'line1\nline2\nline3'

stream=asyncio.StreamReader(loop=self.loop)
stream.feed_data(DATA)
stream.feed_eof()

asyncdefreader():
data= []
asyncforlineinstream:
data.append(line)
returndata

data=self.loop.run_until_complete(reader())
self.assertEqual(data, [b'line1\n', b'line2\n', b'line3'])


classCoroutineTests(BaseTest):

deftest_iscoroutine(self):
asyncdeffoo(): pass

f=foo()
try:
self.assertTrue(asyncio.iscoroutine(f))
finally:
f.close() # silence warning

self.assertTrue(asyncio.iscoroutine(FakeCoro()))

deftest_iscoroutine_generator(self):
deffoo(): yield

self.assertFalse(asyncio.iscoroutine(foo()))


deftest_iscoroutinefunction(self):
asyncdeffoo(): pass
self.assertTrue(asyncio.iscoroutinefunction(foo))

deftest_async_def_coroutines(self):
asyncdefbar():
return'spam'
asyncdeffoo():
returnawaitbar()

# production mode
data=self.loop.run_until_complete(foo())
self.assertEqual(data, 'spam')

# debug mode
self.loop.set_debug(True)
data=self.loop.run_until_complete(foo())
self.assertEqual(data, 'spam')

deftest_debug_mode_manages_coroutine_origin_tracking(self):
asyncdefstart():
self.assertTrue(sys.get_coroutine_origin_tracking_depth() >0)

self.assertEqual(sys.get_coroutine_origin_tracking_depth(), 0)
self.loop.set_debug(True)
self.loop.run_until_complete(start())
self.assertEqual(sys.get_coroutine_origin_tracking_depth(), 0)

deftest_types_coroutine(self):
defgen():
yieldfrom ()
return'spam'

@types.coroutine
deffunc():
returngen()

asyncdefcoro():
wrapper=func()
self.assertIsInstance(wrapper, types._GeneratorWrapper)
returnawaitwrapper

data=self.loop.run_until_complete(coro())
self.assertEqual(data, 'spam')

deftest_task_print_stack(self):
T=None

asyncdeffoo():
f=T.get_stack(limit=1)
try:
self.assertEqual(f[0].f_code.co_name, 'foo')
finally:
f=None

asyncdefrunner():
nonlocalT
T=asyncio.ensure_future(foo(), loop=self.loop)
awaitT

self.loop.run_until_complete(runner())

deftest_double_await(self):
asyncdefafunc():
awaitasyncio.sleep(0.1)

asyncdefrunner():
coro=afunc()
t=self.loop.create_task(coro)
try:
awaitasyncio.sleep(0)
awaitcoro
finally:
t.cancel()

self.loop.set_debug(True)
withself.assertRaises(
RuntimeError,
msg='coroutine is being awaited already'):

self.loop.run_until_complete(runner())


if__name__=='__main__':
unittest.main()