Python code coverage for Lib/test/test_asyncio/test_pep492.py

#	count	content
1	n/a	"""Tests support for new syntax introduced by PEP 492."""
2	n/a
3	n/a	import collections.abc
4	n/a	import types
5	n/a	import unittest
6	n/a
7	n/a	try:
8	n/a	from test import support
9	n/a	except ImportError:
10	n/a	from asyncio import test_support as support
11	n/a	from unittest import mock
12	n/a
13	n/a	import asyncio
14	n/a	from asyncio import test_utils
15	n/a
16	n/a
17	n/a	class BaseTest(test_utils.TestCase):
18	n/a
19	n/a	def setUp(self):
20	n/a	super().setUp()
21	n/a	self.loop = asyncio.BaseEventLoop()
22	n/a	self.loop._process_events = mock.Mock()
23	n/a	self.loop._selector = mock.Mock()
24	n/a	self.loop._selector.select.return_value = ()
25	n/a	self.set_event_loop(self.loop)
26	n/a
27	n/a
28	n/a	class LockTests(BaseTest):
29	n/a
30	n/a	def test_context_manager_async_with(self):
31	n/a	primitives = [
32	n/a	asyncio.Lock(loop=self.loop),
33	n/a	asyncio.Condition(loop=self.loop),
34	n/a	asyncio.Semaphore(loop=self.loop),
35	n/a	asyncio.BoundedSemaphore(loop=self.loop),
36	n/a	]
37	n/a
38	n/a	async def test(lock):
39	n/a	await asyncio.sleep(0.01, loop=self.loop)
40	n/a	self.assertFalse(lock.locked())
41	n/a	async with lock as _lock:
42	n/a	self.assertIs(_lock, None)
43	n/a	self.assertTrue(lock.locked())
44	n/a	await asyncio.sleep(0.01, loop=self.loop)
45	n/a	self.assertTrue(lock.locked())
46	n/a	self.assertFalse(lock.locked())
47	n/a
48	n/a	for primitive in primitives:
49	n/a	self.loop.run_until_complete(test(primitive))
50	n/a	self.assertFalse(primitive.locked())
51	n/a
52	n/a	def test_context_manager_with_await(self):
53	n/a	primitives = [
54	n/a	asyncio.Lock(loop=self.loop),
55	n/a	asyncio.Condition(loop=self.loop),
56	n/a	asyncio.Semaphore(loop=self.loop),
57	n/a	asyncio.BoundedSemaphore(loop=self.loop),
58	n/a	]
59	n/a
60	n/a	async def test(lock):
61	n/a	await asyncio.sleep(0.01, loop=self.loop)
62	n/a	self.assertFalse(lock.locked())
63	n/a	with await lock as _lock:
64	n/a	self.assertIs(_lock, None)
65	n/a	self.assertTrue(lock.locked())
66	n/a	await asyncio.sleep(0.01, loop=self.loop)
67	n/a	self.assertTrue(lock.locked())
68	n/a	self.assertFalse(lock.locked())
69	n/a
70	n/a	for primitive in primitives:
71	n/a	self.loop.run_until_complete(test(primitive))
72	n/a	self.assertFalse(primitive.locked())
73	n/a
74	n/a
75	n/a	class StreamReaderTests(BaseTest):
76	n/a
77	n/a	def test_readline(self):
78	n/a	DATA = b'line1\nline2\nline3'
79	n/a
80	n/a	stream = asyncio.StreamReader(loop=self.loop)
81	n/a	stream.feed_data(DATA)
82	n/a	stream.feed_eof()
83	n/a
84	n/a	async def reader():
85	n/a	data = []
86	n/a	async for line in stream:
87	n/a	data.append(line)
88	n/a	return data
89	n/a
90	n/a	data = self.loop.run_until_complete(reader())
91	n/a	self.assertEqual(data, [b'line1\n', b'line2\n', b'line3'])
92	n/a
93	n/a
94	n/a	class CoroutineTests(BaseTest):
95	n/a
96	n/a	def test_iscoroutine(self):
97	n/a	async def foo(): pass
98	n/a
99	n/a	f = foo()
100	n/a	try:
101	n/a	self.assertTrue(asyncio.iscoroutine(f))
102	n/a	finally:
103	n/a	f.close() # silence warning
104	n/a
105	n/a	# Test that asyncio.iscoroutine() uses collections.abc.Coroutine
106	n/a	class FakeCoro:
107	n/a	def send(self, value): pass
108	n/a	def throw(self, typ, val=None, tb=None): pass
109	n/a	def close(self): pass
110	n/a	def __await__(self): yield
111	n/a
112	n/a	self.assertTrue(asyncio.iscoroutine(FakeCoro()))
113	n/a
114	n/a	def test_iscoroutinefunction(self):
115	n/a	async def foo(): pass
116	n/a	self.assertTrue(asyncio.iscoroutinefunction(foo))
117	n/a
118	n/a	def test_function_returning_awaitable(self):
119	n/a	class Awaitable:
120	n/a	def __await__(self):
121	n/a	return ('spam',)
122	n/a
123	n/a	@asyncio.coroutine
124	n/a	def func():
125	n/a	return Awaitable()
126	n/a
127	n/a	coro = func()
128	n/a	self.assertEqual(coro.send(None), 'spam')
129	n/a	coro.close()
130	n/a
131	n/a	def test_async_def_coroutines(self):
132	n/a	async def bar():
133	n/a	return 'spam'
134	n/a	async def foo():
135	n/a	return await bar()
136	n/a
137	n/a	# production mode
138	n/a	data = self.loop.run_until_complete(foo())
139	n/a	self.assertEqual(data, 'spam')
140	n/a
141	n/a	# debug mode
142	n/a	self.loop.set_debug(True)
143	n/a	data = self.loop.run_until_complete(foo())
144	n/a	self.assertEqual(data, 'spam')
145	n/a
146	n/a	@mock.patch('asyncio.coroutines.logger')
147	n/a	def test_async_def_wrapped(self, m_log):
148	n/a	async def foo():
149	n/a	pass
150	n/a	async def start():
151	n/a	foo_coro = foo()
152	n/a	self.assertRegex(
153	n/a	repr(foo_coro),
154	n/a	r'<CoroWrapper .\.foo\(\) running at .pep492.*>')
155	n/a
156	n/a	with support.check_warnings((r'.foo.was never',
157	n/a	RuntimeWarning)):
158	n/a	foo_coro = None
159	n/a	support.gc_collect()
160	n/a	self.assertTrue(m_log.error.called)
161	n/a	message = m_log.error.call_args[0][0]
162	n/a	self.assertRegex(message,
163	n/a	r'CoroWrapper.foo.was never')
164	n/a
165	n/a	self.loop.set_debug(True)
166	n/a	self.loop.run_until_complete(start())
167	n/a
168	n/a	async def start():
169	n/a	foo_coro = foo()
170	n/a	task = asyncio.ensure_future(foo_coro, loop=self.loop)
171	n/a	self.assertRegex(repr(task), r'Task.foo.running')
172	n/a
173	n/a	self.loop.run_until_complete(start())
174	n/a
175	n/a
176	n/a	def test_types_coroutine(self):
177	n/a	def gen():
178	n/a	yield from ()
179	n/a	return 'spam'
180	n/a
181	n/a	@types.coroutine
182	n/a	def func():
183	n/a	return gen()
184	n/a
185	n/a	async def coro():
186	n/a	wrapper = func()
187	n/a	self.assertIsInstance(wrapper, types._GeneratorWrapper)
188	n/a	return await wrapper
189	n/a
190	n/a	data = self.loop.run_until_complete(coro())
191	n/a	self.assertEqual(data, 'spam')
192	n/a
193	n/a	def test_task_print_stack(self):
194	n/a	T = None
195	n/a
196	n/a	async def foo():
197	n/a	f = T.get_stack(limit=1)
198	n/a	try:
199	n/a	self.assertEqual(f[0].f_code.co_name, 'foo')
200	n/a	finally:
201	n/a	f = None
202	n/a
203	n/a	async def runner():
204	n/a	nonlocal T
205	n/a	T = asyncio.ensure_future(foo(), loop=self.loop)
206	n/a	await T
207	n/a
208	n/a	self.loop.run_until_complete(runner())
209	n/a
210	n/a	def test_double_await(self):
211	n/a	async def afunc():
212	n/a	await asyncio.sleep(0.1, loop=self.loop)
213	n/a
214	n/a	async def runner():
215	n/a	coro = afunc()
216	n/a	t = asyncio.Task(coro, loop=self.loop)
217	n/a	try:
218	n/a	await asyncio.sleep(0, loop=self.loop)
219	n/a	await coro
220	n/a	finally:
221	n/a	t.cancel()
222	n/a
223	n/a	self.loop.set_debug(True)
224	n/a	with self.assertRaisesRegex(
225	n/a	RuntimeError,
226	n/a	r'Cannot await.test_double_await.\bafunc\b.while.\bsleep\b'):
227	n/a
228	n/a	self.loop.run_until_complete(runner())
229	n/a
230	n/a
231	n/a	if __name__ == '__main__':
232	n/a	unittest.main()