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

#	count	content
1	n/a	import os
2	n/a	import sys
3	n/a	import unittest
4	n/a	from unittest import mock
5	n/a
6	n/a	if sys.platform != 'win32':
7	n/a	raise unittest.SkipTest('Windows only')
8	n/a
9	n/a	import _winapi
10	n/a
11	n/a	import asyncio
12	n/a	from asyncio import _overlapped
13	n/a	from asyncio import test_utils
14	n/a	from asyncio import windows_events
15	n/a
16	n/a
17	n/a	class UpperProto(asyncio.Protocol):
18	n/a	def __init__(self):
19	n/a	self.buf = []
20	n/a
21	n/a	def connection_made(self, trans):
22	n/a	self.trans = trans
23	n/a
24	n/a	def data_received(self, data):
25	n/a	self.buf.append(data)
26	n/a	if b'\n' in data:
27	n/a	self.trans.write(b''.join(self.buf).upper())
28	n/a	self.trans.close()
29	n/a
30	n/a
31	n/a	class ProactorTests(test_utils.TestCase):
32	n/a
33	n/a	def setUp(self):
34	n/a	super().setUp()
35	n/a	self.loop = asyncio.ProactorEventLoop()
36	n/a	self.set_event_loop(self.loop)
37	n/a
38	n/a	def test_close(self):
39	n/a	a, b = self.loop._socketpair()
40	n/a	trans = self.loop._make_socket_transport(a, asyncio.Protocol())
41	n/a	f = asyncio.ensure_future(self.loop.sock_recv(b, 100))
42	n/a	trans.close()
43	n/a	self.loop.run_until_complete(f)
44	n/a	self.assertEqual(f.result(), b'')
45	n/a	b.close()
46	n/a
47	n/a	def test_double_bind(self):
48	n/a	ADDRESS = r'\\.\pipe\test_double_bind-%s' % os.getpid()
49	n/a	server1 = windows_events.PipeServer(ADDRESS)
50	n/a	with self.assertRaises(PermissionError):
51	n/a	windows_events.PipeServer(ADDRESS)
52	n/a	server1.close()
53	n/a
54	n/a	def test_pipe(self):
55	n/a	res = self.loop.run_until_complete(self._test_pipe())
56	n/a	self.assertEqual(res, 'done')
57	n/a
58	n/a	def _test_pipe(self):
59	n/a	ADDRESS = r'\\.\pipe\_test_pipe-%s' % os.getpid()
60	n/a
61	n/a	with self.assertRaises(FileNotFoundError):
62	n/a	yield from self.loop.create_pipe_connection(
63	n/a	asyncio.Protocol, ADDRESS)
64	n/a
65	n/a	[server] = yield from self.loop.start_serving_pipe(
66	n/a	UpperProto, ADDRESS)
67	n/a	self.assertIsInstance(server, windows_events.PipeServer)
68	n/a
69	n/a	clients = []
70	n/a	for i in range(5):
71	n/a	stream_reader = asyncio.StreamReader(loop=self.loop)
72	n/a	protocol = asyncio.StreamReaderProtocol(stream_reader,
73	n/a	loop=self.loop)
74	n/a	trans, proto = yield from self.loop.create_pipe_connection(
75	n/a	lambda: protocol, ADDRESS)
76	n/a	self.assertIsInstance(trans, asyncio.Transport)
77	n/a	self.assertEqual(protocol, proto)
78	n/a	clients.append((stream_reader, trans))
79	n/a
80	n/a	for i, (r, w) in enumerate(clients):
81	n/a	w.write('lower-{}\n'.format(i).encode())
82	n/a
83	n/a	for i, (r, w) in enumerate(clients):
84	n/a	response = yield from r.readline()
85	n/a	self.assertEqual(response, 'LOWER-{}\n'.format(i).encode())
86	n/a	w.close()
87	n/a
88	n/a	server.close()
89	n/a
90	n/a	with self.assertRaises(FileNotFoundError):
91	n/a	yield from self.loop.create_pipe_connection(
92	n/a	asyncio.Protocol, ADDRESS)
93	n/a
94	n/a	return 'done'
95	n/a
96	n/a	def test_connect_pipe_cancel(self):
97	n/a	exc = OSError()
98	n/a	exc.winerror = _overlapped.ERROR_PIPE_BUSY
99	n/a	with mock.patch.object(_overlapped, 'ConnectPipe', side_effect=exc) as connect:
100	n/a	coro = self.loop._proactor.connect_pipe('pipe_address')
101	n/a	task = self.loop.create_task(coro)
102	n/a
103	n/a	# check that it's possible to cancel connect_pipe()
104	n/a	task.cancel()
105	n/a	with self.assertRaises(asyncio.CancelledError):
106	n/a	self.loop.run_until_complete(task)
107	n/a
108	n/a	def test_wait_for_handle(self):
109	n/a	event = _overlapped.CreateEvent(None, True, False, None)
110	n/a	self.addCleanup(_winapi.CloseHandle, event)
111	n/a
112	n/a	# Wait for unset event with 0.5s timeout;
113	n/a	# result should be False at timeout
114	n/a	fut = self.loop._proactor.wait_for_handle(event, 0.5)
115	n/a	start = self.loop.time()
116	n/a	done = self.loop.run_until_complete(fut)
117	n/a	elapsed = self.loop.time() - start
118	n/a
119	n/a	self.assertEqual(done, False)
120	n/a	self.assertFalse(fut.result())
121	n/a	self.assertTrue(0.48 < elapsed < 0.9, elapsed)
122	n/a
123	n/a	_overlapped.SetEvent(event)
124	n/a
125	n/a	# Wait for set event;
126	n/a	# result should be True immediately
127	n/a	fut = self.loop._proactor.wait_for_handle(event, 10)
128	n/a	start = self.loop.time()
129	n/a	done = self.loop.run_until_complete(fut)
130	n/a	elapsed = self.loop.time() - start
131	n/a
132	n/a	self.assertEqual(done, True)
133	n/a	self.assertTrue(fut.result())
134	n/a	self.assertTrue(0 <= elapsed < 0.3, elapsed)
135	n/a
136	n/a	# asyncio issue #195: cancelling a done _WaitHandleFuture
137	n/a	# must not crash
138	n/a	fut.cancel()
139	n/a
140	n/a	def test_wait_for_handle_cancel(self):
141	n/a	event = _overlapped.CreateEvent(None, True, False, None)
142	n/a	self.addCleanup(_winapi.CloseHandle, event)
143	n/a
144	n/a	# Wait for unset event with a cancelled future;
145	n/a	# CancelledError should be raised immediately
146	n/a	fut = self.loop._proactor.wait_for_handle(event, 10)
147	n/a	fut.cancel()
148	n/a	start = self.loop.time()
149	n/a	with self.assertRaises(asyncio.CancelledError):
150	n/a	self.loop.run_until_complete(fut)
151	n/a	elapsed = self.loop.time() - start
152	n/a	self.assertTrue(0 <= elapsed < 0.1, elapsed)
153	n/a
154	n/a	# asyncio issue #195: cancelling a _WaitHandleFuture twice
155	n/a	# must not crash
156	n/a	fut = self.loop._proactor.wait_for_handle(event)
157	n/a	fut.cancel()
158	n/a	fut.cancel()
159	n/a
160	n/a
161	n/a	if __name__ == '__main__':
162	n/a	unittest.main()