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

#	count	content
1	n/a	"""Tests for window_utils"""
2	n/a
3	n/a	import socket
4	n/a	import sys
5	n/a	import unittest
6	n/a	import warnings
7	n/a	from unittest import mock
8	n/a
9	n/a	if sys.platform != 'win32':
10	n/a	raise unittest.SkipTest('Windows only')
11	n/a
12	n/a	import _winapi
13	n/a
14	n/a	from asyncio import _overlapped
15	n/a	from asyncio import windows_utils
16	n/a	try:
17	n/a	from test import support
18	n/a	except ImportError:
19	n/a	from asyncio import test_support as support
20	n/a
21	n/a
22	n/a	class WinsocketpairTests(unittest.TestCase):
23	n/a
24	n/a	def check_winsocketpair(self, ssock, csock):
25	n/a	csock.send(b'xxx')
26	n/a	self.assertEqual(b'xxx', ssock.recv(1024))
27	n/a	csock.close()
28	n/a	ssock.close()
29	n/a
30	n/a	def test_winsocketpair(self):
31	n/a	ssock, csock = windows_utils.socketpair()
32	n/a	self.check_winsocketpair(ssock, csock)
33	n/a
34	n/a	@unittest.skipUnless(support.IPV6_ENABLED, 'IPv6 not supported or enabled')
35	n/a	def test_winsocketpair_ipv6(self):
36	n/a	ssock, csock = windows_utils.socketpair(family=socket.AF_INET6)
37	n/a	self.check_winsocketpair(ssock, csock)
38	n/a
39	n/a	@unittest.skipIf(hasattr(socket, 'socketpair'),
40	n/a	'socket.socketpair is available')
41	n/a	@mock.patch('asyncio.windows_utils.socket')
42	n/a	def test_winsocketpair_exc(self, m_socket):
43	n/a	m_socket.AF_INET = socket.AF_INET
44	n/a	m_socket.SOCK_STREAM = socket.SOCK_STREAM
45	n/a	m_socket.socket.return_value.getsockname.return_value = ('', 12345)
46	n/a	m_socket.socket.return_value.accept.return_value = object(), object()
47	n/a	m_socket.socket.return_value.connect.side_effect = OSError()
48	n/a
49	n/a	self.assertRaises(OSError, windows_utils.socketpair)
50	n/a
51	n/a	def test_winsocketpair_invalid_args(self):
52	n/a	self.assertRaises(ValueError,
53	n/a	windows_utils.socketpair, family=socket.AF_UNSPEC)
54	n/a	self.assertRaises(ValueError,
55	n/a	windows_utils.socketpair, type=socket.SOCK_DGRAM)
56	n/a	self.assertRaises(ValueError,
57	n/a	windows_utils.socketpair, proto=1)
58	n/a
59	n/a	@unittest.skipIf(hasattr(socket, 'socketpair'),
60	n/a	'socket.socketpair is available')
61	n/a	@mock.patch('asyncio.windows_utils.socket')
62	n/a	def test_winsocketpair_close(self, m_socket):
63	n/a	m_socket.AF_INET = socket.AF_INET
64	n/a	m_socket.SOCK_STREAM = socket.SOCK_STREAM
65	n/a	sock = mock.Mock()
66	n/a	m_socket.socket.return_value = sock
67	n/a	sock.bind.side_effect = OSError
68	n/a	self.assertRaises(OSError, windows_utils.socketpair)
69	n/a	self.assertTrue(sock.close.called)
70	n/a
71	n/a
72	n/a	class PipeTests(unittest.TestCase):
73	n/a
74	n/a	def test_pipe_overlapped(self):
75	n/a	h1, h2 = windows_utils.pipe(overlapped=(True, True))
76	n/a	try:
77	n/a	ov1 = _overlapped.Overlapped()
78	n/a	self.assertFalse(ov1.pending)
79	n/a	self.assertEqual(ov1.error, 0)
80	n/a
81	n/a	ov1.ReadFile(h1, 100)
82	n/a	self.assertTrue(ov1.pending)
83	n/a	self.assertEqual(ov1.error, _winapi.ERROR_IO_PENDING)
84	n/a	ERROR_IO_INCOMPLETE = 996
85	n/a	try:
86	n/a	ov1.getresult()
87	n/a	except OSError as e:
88	n/a	self.assertEqual(e.winerror, ERROR_IO_INCOMPLETE)
89	n/a	else:
90	n/a	raise RuntimeError('expected ERROR_IO_INCOMPLETE')
91	n/a
92	n/a	ov2 = _overlapped.Overlapped()
93	n/a	self.assertFalse(ov2.pending)
94	n/a	self.assertEqual(ov2.error, 0)
95	n/a
96	n/a	ov2.WriteFile(h2, b"hello")
97	n/a	self.assertIn(ov2.error, {0, _winapi.ERROR_IO_PENDING})
98	n/a
99	n/a	res = _winapi.WaitForMultipleObjects([ov2.event], False, 100)
100	n/a	self.assertEqual(res, _winapi.WAIT_OBJECT_0)
101	n/a
102	n/a	self.assertFalse(ov1.pending)
103	n/a	self.assertEqual(ov1.error, ERROR_IO_INCOMPLETE)
104	n/a	self.assertFalse(ov2.pending)
105	n/a	self.assertIn(ov2.error, {0, _winapi.ERROR_IO_PENDING})
106	n/a	self.assertEqual(ov1.getresult(), b"hello")
107	n/a	finally:
108	n/a	_winapi.CloseHandle(h1)
109	n/a	_winapi.CloseHandle(h2)
110	n/a
111	n/a	def test_pipe_handle(self):
112	n/a	h, _ = windows_utils.pipe(overlapped=(True, True))
113	n/a	_winapi.CloseHandle(_)
114	n/a	p = windows_utils.PipeHandle(h)
115	n/a	self.assertEqual(p.fileno(), h)
116	n/a	self.assertEqual(p.handle, h)
117	n/a
118	n/a	# check garbage collection of p closes handle
119	n/a	with warnings.catch_warnings():
120	n/a	warnings.filterwarnings("ignore", "", ResourceWarning)
121	n/a	del p
122	n/a	support.gc_collect()
123	n/a	try:
124	n/a	_winapi.CloseHandle(h)
125	n/a	except OSError as e:
126	n/a	self.assertEqual(e.winerror, 6) # ERROR_INVALID_HANDLE
127	n/a	else:
128	n/a	raise RuntimeError('expected ERROR_INVALID_HANDLE')
129	n/a
130	n/a
131	n/a	class PopenTests(unittest.TestCase):
132	n/a
133	n/a	def test_popen(self):
134	n/a	command = r"""if 1:
135	n/a	import sys
136	n/a	s = sys.stdin.readline()
137	n/a	sys.stdout.write(s.upper())
138	n/a	sys.stderr.write('stderr')
139	n/a	"""
140	n/a	msg = b"blah\n"
141	n/a
142	n/a	p = windows_utils.Popen([sys.executable, '-c', command],
143	n/a	stdin=windows_utils.PIPE,
144	n/a	stdout=windows_utils.PIPE,
145	n/a	stderr=windows_utils.PIPE)
146	n/a
147	n/a	for f in [p.stdin, p.stdout, p.stderr]:
148	n/a	self.assertIsInstance(f, windows_utils.PipeHandle)
149	n/a
150	n/a	ovin = _overlapped.Overlapped()
151	n/a	ovout = _overlapped.Overlapped()
152	n/a	overr = _overlapped.Overlapped()
153	n/a
154	n/a	ovin.WriteFile(p.stdin.handle, msg)
155	n/a	ovout.ReadFile(p.stdout.handle, 100)
156	n/a	overr.ReadFile(p.stderr.handle, 100)
157	n/a
158	n/a	events = [ovin.event, ovout.event, overr.event]
159	n/a	# Super-long timeout for slow buildbots.
160	n/a	res = _winapi.WaitForMultipleObjects(events, True, 10000)
161	n/a	self.assertEqual(res, _winapi.WAIT_OBJECT_0)
162	n/a	self.assertFalse(ovout.pending)
163	n/a	self.assertFalse(overr.pending)
164	n/a	self.assertFalse(ovin.pending)
165	n/a
166	n/a	self.assertEqual(ovin.getresult(), len(msg))
167	n/a	out = ovout.getresult().rstrip()
168	n/a	err = overr.getresult().rstrip()
169	n/a
170	n/a	self.assertGreater(len(out), 0)
171	n/a	self.assertGreater(len(err), 0)
172	n/a	# allow for partial reads...
173	n/a	self.assertTrue(msg.upper().rstrip().startswith(out))
174	n/a	self.assertTrue(b"stderr".startswith(err))
175	n/a
176	n/a	# The context manager calls wait() and closes resources
177	n/a	with p:
178	n/a	pass
179	n/a
180	n/a
181	n/a	if __name__ == '__main__':
182	n/a	unittest.main()