(self)
| 81 | class 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.assertStartsWith(msg.upper().rstrip(), out) |
| 125 | self.assertStartsWith(b"stderr", err) |
| 126 | |
| 127 | # The context manager calls wait() and closes resources |
| 128 | with p: |
| 129 | pass |
| 130 | |
| 131 | |
| 132 | class OverlappedRefleakTests(unittest.TestCase): |
nothing calls this directly
no test coverage detected