summaryrefslogtreecommitdiffstats
path: root/Lib
diff options
context:
space:
mode:
authorGregory P. Smith ext:(%20%5BGoogle%20Inc.%5D) <greg@krypto.org>2016-06-05 02:57:47 (GMT)
committerGregory P. Smith ext:(%20%5BGoogle%20Inc.%5D) <greg@krypto.org>2016-06-05 02:57:47 (GMT)
commit2daf8e7f76c2b6917468dc50ebae792f34481cf4 (patch)
treee4e9f7be6747c016902c0fb5d5bcf50c10779c1e /Lib
parent92b4b5a863bcdc09150bf3668092e4ceb7abec44 (diff)
downloadcpython-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.py92
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):