ยปCore Development>Code coverage>Lib/test/test_asyncio/test_windows_utils.py

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

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