• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1"""Tests for window_utils"""
2
3import sys
4import unittest
5import warnings
6
7if sys.platform != 'win32':
8    raise unittest.SkipTest('Windows only')
9
10import _overlapped
11import _winapi
12
13import asyncio
14from asyncio import windows_utils
15from test import support
16
17
18def tearDownModule():
19    asyncio.set_event_loop_policy(None)
20
21
22class PipeTests(unittest.TestCase):
23
24    def test_pipe_overlapped(self):
25        h1, h2 = windows_utils.pipe(overlapped=(True, True))
26        try:
27            ov1 = _overlapped.Overlapped()
28            self.assertFalse(ov1.pending)
29            self.assertEqual(ov1.error, 0)
30
31            ov1.ReadFile(h1, 100)
32            self.assertTrue(ov1.pending)
33            self.assertEqual(ov1.error, _winapi.ERROR_IO_PENDING)
34            ERROR_IO_INCOMPLETE = 996
35            try:
36                ov1.getresult()
37            except OSError as e:
38                self.assertEqual(e.winerror, ERROR_IO_INCOMPLETE)
39            else:
40                raise RuntimeError('expected ERROR_IO_INCOMPLETE')
41
42            ov2 = _overlapped.Overlapped()
43            self.assertFalse(ov2.pending)
44            self.assertEqual(ov2.error, 0)
45
46            ov2.WriteFile(h2, b"hello")
47            self.assertIn(ov2.error, {0, _winapi.ERROR_IO_PENDING})
48
49            res = _winapi.WaitForMultipleObjects([ov2.event], False, 100)
50            self.assertEqual(res, _winapi.WAIT_OBJECT_0)
51
52            self.assertFalse(ov1.pending)
53            self.assertEqual(ov1.error, ERROR_IO_INCOMPLETE)
54            self.assertFalse(ov2.pending)
55            self.assertIn(ov2.error, {0, _winapi.ERROR_IO_PENDING})
56            self.assertEqual(ov1.getresult(), b"hello")
57        finally:
58            _winapi.CloseHandle(h1)
59            _winapi.CloseHandle(h2)
60
61    def test_pipe_handle(self):
62        h, _ = windows_utils.pipe(overlapped=(True, True))
63        _winapi.CloseHandle(_)
64        p = windows_utils.PipeHandle(h)
65        self.assertEqual(p.fileno(), h)
66        self.assertEqual(p.handle, h)
67
68        # check garbage collection of p closes handle
69        with warnings.catch_warnings():
70            warnings.filterwarnings("ignore", "",  ResourceWarning)
71            del p
72            support.gc_collect()
73        try:
74            _winapi.CloseHandle(h)
75        except OSError as e:
76            self.assertEqual(e.winerror, 6)     # ERROR_INVALID_HANDLE
77        else:
78            raise RuntimeError('expected ERROR_INVALID_HANDLE')
79
80
81class PopenTests(unittest.TestCase):
82
83    def test_popen(self):
84        command = r"""if 1:
85            import sys
86            s = sys.stdin.readline()
87            sys.stdout.write(s.upper())
88            sys.stderr.write('stderr')
89            """
90        msg = b"blah\n"
91
92        p = windows_utils.Popen([sys.executable, '-c', command],
93                                stdin=windows_utils.PIPE,
94                                stdout=windows_utils.PIPE,
95                                stderr=windows_utils.PIPE)
96
97        for f in [p.stdin, p.stdout, p.stderr]:
98            self.assertIsInstance(f, windows_utils.PipeHandle)
99
100        ovin = _overlapped.Overlapped()
101        ovout = _overlapped.Overlapped()
102        overr = _overlapped.Overlapped()
103
104        ovin.WriteFile(p.stdin.handle, msg)
105        ovout.ReadFile(p.stdout.handle, 100)
106        overr.ReadFile(p.stderr.handle, 100)
107
108        events = [ovin.event, ovout.event, overr.event]
109        # Super-long timeout for slow buildbots.
110        res = _winapi.WaitForMultipleObjects(events, True,
111                                             int(support.SHORT_TIMEOUT * 1000))
112        self.assertEqual(res, _winapi.WAIT_OBJECT_0)
113        self.assertFalse(ovout.pending)
114        self.assertFalse(overr.pending)
115        self.assertFalse(ovin.pending)
116
117        self.assertEqual(ovin.getresult(), len(msg))
118        out = ovout.getresult().rstrip()
119        err = overr.getresult().rstrip()
120
121        self.assertGreater(len(out), 0)
122        self.assertGreater(len(err), 0)
123        # allow for partial reads...
124        self.assertTrue(msg.upper().rstrip().startswith(out))
125        self.assertTrue(b"stderr".startswith(err))
126
127        # The context manager calls wait() and closes resources
128        with p:
129            pass
130
131
132if __name__ == '__main__':
133    unittest.main()
134