(original) (raw)

changeset: 101737:e89eb7935ca9 parent: 101735:edf5ae3162ef parent: 101736:97bdba07e361 user: Gregory P. Smith greg@krypto.org [Google Inc.] date: Sun Jun 05 02:58:38 2016 +0000 files: Lib/test/test_subprocess.py description: merge from 3.5. (moves the issue26372 tests to the proper class) diff -r edf5ae3162ef -r e89eb7935ca9 Lib/test/test_subprocess.py --- a/Lib/test/test_subprocess.py Sat Jun 04 17:24:01 2016 -0700 +++ b/Lib/test/test_subprocess.py Sun Jun 05 02:58:38 2016 +0000 @@ -1238,52 +1238,6 @@ fds_after_exception = os.listdir(fd_directory) self.assertEqual(fds_before_popen, fds_after_exception) - def test_communicate_BrokenPipeError_stdin_close(self): - # By not setting stdout or stderr or a timeout we force the fast path - # that just calls _stdin_write() internally due to our mock. - proc = subprocess.Popen([sys.executable, '-c', 'pass']) - with proc, mock.patch.object(proc, 'stdin') as mock_proc_stdin: - mock_proc_stdin.close.side_effect = BrokenPipeError - proc.communicate() # Should swallow BrokenPipeError from close. - mock_proc_stdin.close.assert_called_with() - - def test_communicate_BrokenPipeError_stdin_write(self): - # By not setting stdout or stderr or a timeout we force the fast path - # that just calls _stdin_write() internally due to our mock. - proc = subprocess.Popen([sys.executable, '-c', 'pass']) - with proc, mock.patch.object(proc, 'stdin') as mock_proc_stdin: - mock_proc_stdin.write.side_effect = BrokenPipeError - proc.communicate(b'stuff') # Should swallow the BrokenPipeError. - mock_proc_stdin.write.assert_called_once_with(b'stuff') - mock_proc_stdin.close.assert_called_once_with() - - def test_communicate_BrokenPipeError_stdin_flush(self): - # Setting stdin and stdout forces the ._communicate() code path. - # python -h exits faster than python -c pass (but spams stdout). - proc = subprocess.Popen([sys.executable, '-h'], - stdin=subprocess.PIPE, - stdout=subprocess.PIPE) - with proc, mock.patch.object(proc, 'stdin') as mock_proc_stdin, \ - open(os.devnull, 'wb') as dev_null: - mock_proc_stdin.flush.side_effect = BrokenPipeError - # because _communicate registers a selector using proc.stdin... - mock_proc_stdin.fileno.return_value = dev_null.fileno() - # _communicate() should swallow BrokenPipeError from flush. - proc.communicate(b'stuff') - mock_proc_stdin.flush.assert_called_once_with() - - def test_communicate_BrokenPipeError_stdin_close_with_timeout(self): - # Setting stdin and stdout forces the ._communicate() code path. - # python -h exits faster than python -c pass (but spams stdout). - proc = subprocess.Popen([sys.executable, '-h'], - stdin=subprocess.PIPE, - stdout=subprocess.PIPE) - with proc, mock.patch.object(proc, 'stdin') as mock_proc_stdin: - mock_proc_stdin.close.side_effect = BrokenPipeError - # _communicate() should swallow BrokenPipeError from close. - proc.communicate(timeout=999) - mock_proc_stdin.close.assert_called_once_with() - class RunFuncTestCase(BaseTestCase): def run_python(self, code, **kwargs): @@ -2470,6 +2424,52 @@ if not gc_enabled: gc.disable() + def test_communicate_BrokenPipeError_stdin_close(self): + # By not setting stdout or stderr or a timeout we force the fast path + # that just calls _stdin_write() internally due to our mock. + proc = subprocess.Popen([sys.executable, '-c', 'pass']) + with proc, mock.patch.object(proc, 'stdin') as mock_proc_stdin: + mock_proc_stdin.close.side_effect = BrokenPipeError + proc.communicate() # Should swallow BrokenPipeError from close. + mock_proc_stdin.close.assert_called_with() + + def test_communicate_BrokenPipeError_stdin_write(self): + # By not setting stdout or stderr or a timeout we force the fast path + # that just calls _stdin_write() internally due to our mock. + proc = subprocess.Popen([sys.executable, '-c', 'pass']) + with proc, mock.patch.object(proc, 'stdin') as mock_proc_stdin: + mock_proc_stdin.write.side_effect = BrokenPipeError + proc.communicate(b'stuff') # Should swallow the BrokenPipeError. + mock_proc_stdin.write.assert_called_once_with(b'stuff') + mock_proc_stdin.close.assert_called_once_with() + + def test_communicate_BrokenPipeError_stdin_flush(self): + # Setting stdin and stdout forces the ._communicate() code path. + # python -h exits faster than python -c pass (but spams stdout). + proc = subprocess.Popen([sys.executable, '-h'], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE) + with proc, mock.patch.object(proc, 'stdin') as mock_proc_stdin, \ + open(os.devnull, 'wb') as dev_null: + mock_proc_stdin.flush.side_effect = BrokenPipeError + # because _communicate registers a selector using proc.stdin... + mock_proc_stdin.fileno.return_value = dev_null.fileno() + # _communicate() should swallow BrokenPipeError from flush. + proc.communicate(b'stuff') + mock_proc_stdin.flush.assert_called_once_with() + + def test_communicate_BrokenPipeError_stdin_close_with_timeout(self): + # Setting stdin and stdout forces the ._communicate() code path. + # python -h exits faster than python -c pass (but spams stdout). + proc = subprocess.Popen([sys.executable, '-h'], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE) + with proc, mock.patch.object(proc, 'stdin') as mock_proc_stdin: + mock_proc_stdin.close.side_effect = BrokenPipeError + # _communicate() should swallow BrokenPipeError from close. + proc.communicate(timeout=999) + mock_proc_stdin.close.assert_called_once_with() + @unittest.skipUnless(mswindows, "Windows specific tests") class Win32ProcessTestCase(BaseTestCase): /greg@krypto.org