diff options
author | Gregory P. Smith ext:(%20%5BGoogle%20Inc.%5D) <greg@krypto.org> | 2016-06-05 02:57:47 (GMT) |
---|---|---|
committer | Gregory P. Smith ext:(%20%5BGoogle%20Inc.%5D) <greg@krypto.org> | 2016-06-05 02:57:47 (GMT) |
commit | 2daf8e7f76c2b6917468dc50ebae792f34481cf4 (patch) | |
tree | e4e9f7be6747c016902c0fb5d5bcf50c10779c1e /Lib | |
parent | 92b4b5a863bcdc09150bf3668092e4ceb7abec44 (diff) | |
download | cpython-2daf8e7f76c2b6917468dc50ebae792f34481cf4.zip cpython-2daf8e7f76c2b6917468dc50ebae792f34481cf4.tar.gz cpython-2daf8e7f76c2b6917468dc50ebae792f34481cf4.tar.bz2 |
Move the BrokenPipeError tests to the POSIXProcessTestCase class
instead of the generic ProcessTestCase class as they are posix
specific.
Diffstat (limited to 'Lib')
-rw-r--r-- | Lib/test/test_subprocess.py | 92 |
1 files changed, 46 insertions, 46 deletions
diff --git a/Lib/test/test_subprocess.py b/Lib/test/test_subprocess.py index 9e44ee3..4704d49 100644 --- a/Lib/test/test_subprocess.py +++ b/Lib/test/test_subprocess.py @@ -1241,52 +1241,6 @@ class ProcessTestCase(BaseTestCase): 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): @@ -2448,6 +2402,52 @@ class POSIXProcessTestCase(BaseTestCase): 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): |