with _PopenSelector() as selector:
if self.stdin and input:
selector.register(self.stdin, selectors.EVENT_WRITE)
- if self.stdout:
+ if self.stdout and not self.stdout.closed:
selector.register(self.stdout, selectors.EVENT_READ)
- if self.stderr:
+ if self.stderr and not self.stderr.closed:
selector.register(self.stderr, selectors.EVENT_READ)
while selector.get_map():
self.assertEqual(returncode, -3)
+ def test_communicate_repeated_call_after_stdout_close(self):
+ proc = subprocess.Popen([sys.executable, '-c',
+ 'import os, time; os.close(1), time.sleep(2)'],
+ stdout=subprocess.PIPE)
+ while True:
+ try:
+ proc.communicate(timeout=0.1)
+ return
+ except subprocess.TimeoutExpired:
+ pass
+
@unittest.skipUnless(mswindows, "Windows specific tests")
class Win32ProcessTestCase(BaseTestCase):
--- /dev/null
+Fixed :func:`Popen.communicate` subsequent call crash when the child process
+has already closed any piped standard stream, but still continues to be
+running. Patch by Andriy Maletsky.