(self)
| 1122 | self.assertEqual(stderr, None) |
| 1123 | |
| 1124 | def test_communicate_pipe_buf(self): |
| 1125 | # communicate() with writes larger than pipe_buf |
| 1126 | # This test will probably deadlock rather than fail, if |
| 1127 | # communicate() does not work properly. |
| 1128 | x, y = os.pipe() |
| 1129 | os.close(x) |
| 1130 | os.close(y) |
| 1131 | p = subprocess.Popen([sys.executable, "-c", |
| 1132 | 'import sys,os;' |
| 1133 | 'sys.stdout.write(sys.stdin.read(47));' |
| 1134 | 'sys.stderr.write("x" * %d);' |
| 1135 | 'sys.stdout.write(sys.stdin.read())' % |
| 1136 | support.PIPE_MAX_SIZE], |
| 1137 | stdin=subprocess.PIPE, |
| 1138 | stdout=subprocess.PIPE, |
| 1139 | stderr=subprocess.PIPE) |
| 1140 | self.addCleanup(p.stdout.close) |
| 1141 | self.addCleanup(p.stderr.close) |
| 1142 | self.addCleanup(p.stdin.close) |
| 1143 | string_to_write = b"a" * support.PIPE_MAX_SIZE |
| 1144 | (stdout, stderr) = p.communicate(string_to_write) |
| 1145 | self.assertEqual(stdout, string_to_write) |
| 1146 | |
| 1147 | def test_writes_before_communicate(self): |
| 1148 | # stdin.write before communicate() |
nothing calls this directly
no test coverage detected