Index: tests/subprocess2_test.py |
diff --git a/tests/subprocess2_test.py b/tests/subprocess2_test.py |
index 10290592d77cfcd2fefa2ab03bf83f8d429b4936..55936e5582bac65d1583dae2c9ca21777b6dcc91 100755 |
--- a/tests/subprocess2_test.py |
+++ b/tests/subprocess2_test.py |
@@ -28,6 +28,14 @@ from testing_support import auto_stub |
# pylint: disable=R0201 |
+# Create aliases for subprocess2 specific tests. They shouldn't be used for |
+# regression tests. |
+TIMED_OUT = subprocess2.TIMED_OUT |
+VOID = subprocess2.VOID |
+PIPE = subprocess2.PIPE |
+STDOUT = subprocess2.STDOUT |
+ |
+ |
def convert_to_crlf(string): |
"""Unconditionally convert LF to CRLF.""" |
return string.replace('\n', '\r\n') |
@@ -202,8 +210,9 @@ class RegressionTest(BaseTestCase): |
function(noop, self.exe + ['--cr'], True, subp) |
function(noop, self.exe + ['--crlf'], True, subp) |
- def _check_pipes(self, subp, e, stdout, stderr): |
+ def _check_exception(self, subp, e, stdout, stderr, returncode): |
"""On exception, look if the exception members are set correctly.""" |
+ self.assertEquals(returncode, e.returncode) |
if subp is subprocess: |
# subprocess never save the output. |
self.assertFalse(hasattr(e, 'stdout')) |
@@ -239,8 +248,7 @@ class RegressionTest(BaseTestCase): |
e + ['--fail', '--stdout'], universal_newlines=un) |
self.fail() |
except subp.CalledProcessError, e: |
- self._check_pipes(subp, e, c('A\nBB\nCCC\n'), None) |
- self.assertEquals(64, e.returncode) |
+ self._check_exception(subp, e, c('A\nBB\nCCC\n'), None, 64) |
self._run_test(fn) |
def test_check_output_throw_no_stderr(self): |
@@ -252,8 +260,7 @@ class RegressionTest(BaseTestCase): |
e + ['--fail', '--stderr'], universal_newlines=un) |
self.fail() |
except subp.CalledProcessError, e: |
- self._check_pipes(subp, e, c(''), None) |
- self.assertEquals(64, e.returncode) |
+ self._check_exception(subp, e, c(''), None, 64) |
self._run_test(fn) |
def test_check_output_throw_stderr(self): |
@@ -267,8 +274,7 @@ class RegressionTest(BaseTestCase): |
universal_newlines=un) |
self.fail() |
except subp.CalledProcessError, e: |
- self._check_pipes(subp, e, '', c('a\nbb\nccc\n')) |
- self.assertEquals(64, e.returncode) |
+ self._check_exception(subp, e, '', c('a\nbb\nccc\n'), 64) |
self._run_test(fn) |
def test_check_output_throw_stderr_stdout(self): |
@@ -282,8 +288,7 @@ class RegressionTest(BaseTestCase): |
universal_newlines=un) |
self.fail() |
except subp.CalledProcessError, e: |
- self._check_pipes(subp, e, c('a\nbb\nccc\n'), None) |
- self.assertEquals(64, e.returncode) |
+ self._check_exception(subp, e, c('a\nbb\nccc\n'), None, 64) |
self._run_test(fn) |
def test_check_call_throw(self): |
@@ -292,8 +297,7 @@ class RegressionTest(BaseTestCase): |
subp.check_call(self.exe + ['--fail', '--stderr']) |
self.fail() |
except subp.CalledProcessError, e: |
- self._check_pipes(subp, e, None, None) |
- self.assertEquals(64, e.returncode) |
+ self._check_exception(subp, e, None, None, 64) |
class S2Test(BaseTestCase): |
@@ -322,65 +326,80 @@ class S2Test(BaseTestCase): |
function(noop, self.exe + ['--cr'], True) |
function(noop, self.exe + ['--crlf'], True) |
+ def _check_res(self, res, stdout, stderr, returncode): |
+ (out, err), code = res |
+ self.assertEquals(stdout, out) |
+ self.assertEquals(stderr, err) |
+ self.assertEquals(returncode, code) |
+ |
def test_timeout(self): |
# timeout doesn't exist in subprocess. |
def fn(c, e, un): |
- out, returncode = subprocess2.communicate( |
+ res = subprocess2.communicate( |
self.exe + ['--sleep_first', '--stdout'], |
timeout=0.01, |
- stdout=subprocess2.PIPE, |
+ stdout=PIPE, |
shell=False) |
- self.assertEquals(subprocess2.TIMED_OUT, returncode) |
- self.assertEquals(('', None), out) |
+ self._check_res(res, '', None, TIMED_OUT) |
+ self._run_test(fn) |
+ |
+ def test_timeout_shell_throws(self): |
+ def fn(c, e, un): |
+ try: |
+ # With shell=True, it needs a string. |
+ subprocess2.communicate(' '.join(self.exe), timeout=0.01, shell=True) |
+ self.fail() |
+ except TypeError: |
+ pass |
self._run_test(fn) |
def test_stdout_void(self): |
def fn(c, e, un): |
- (out, err), code = subprocess2.communicate( |
+ res = subprocess2.communicate( |
e + ['--stdout', '--stderr'], |
- stdout=subprocess2.VOID, |
- stderr=subprocess2.PIPE, |
+ stdout=VOID, |
+ stderr=PIPE, |
universal_newlines=un) |
- self.assertEquals(None, out) |
- self.assertEquals(c('a\nbb\nccc\n'), err) |
- self.assertEquals(0, code) |
+ self._check_res(res, None, c('a\nbb\nccc\n'), 0) |
self._run_test(fn) |
def test_stderr_void(self): |
def fn(c, e, un): |
- (out, err), code = subprocess2.communicate( |
+ res = subprocess2.communicate( |
e + ['--stdout', '--stderr'], |
- stdout=subprocess2.PIPE, |
- stderr=subprocess2.VOID, |
+ stdout=PIPE, |
+ stderr=VOID, |
universal_newlines=un) |
- self.assertEquals(c('A\nBB\nCCC\n'), out) |
- self.assertEquals(None, err) |
- self.assertEquals(0, code) |
+ self._check_res(res, c('A\nBB\nCCC\n'), None, 0) |
+ self._run_test(fn) |
+ |
+ def test_stdout_void_stderr_redirect(self): |
+ def fn(c, e, un): |
+ res = subprocess2.communicate( |
+ e + ['--stdout', '--stderr'], |
+ stdout=VOID, |
+ stderr=STDOUT, |
+ universal_newlines=un) |
+ self._check_res(res, None, None, 0) |
self._run_test(fn) |
def test_check_output_redirect_stderr_to_stdout_pipe(self): |
def fn(c, e, un): |
- (out, err), code = subprocess2.communicate( |
+ # stderr output into stdout. |
+ res = subprocess2.communicate( |
e + ['--stderr'], |
- stdout=subprocess2.PIPE, |
- stderr=subprocess2.STDOUT, |
+ stdout=PIPE, |
+ stderr=STDOUT, |
universal_newlines=un) |
- # stderr output into stdout. |
- self.assertEquals(c('a\nbb\nccc\n'), out) |
- self.assertEquals(None, err) |
- self.assertEquals(0, code) |
+ self._check_res(res, c('a\nbb\nccc\n'), None, 0) |
self._run_test(fn) |
def test_check_output_redirect_stderr_to_stdout(self): |
def fn(c, e, un): |
- (out, err), code = subprocess2.communicate( |
- e + ['--stderr'], |
- stderr=subprocess2.STDOUT, |
- universal_newlines=un) |
# stderr output into stdout but stdout is not piped. |
- self.assertEquals(None, out) |
- self.assertEquals(None, err) |
- self.assertEquals(0, code) |
+ res = subprocess2.communicate( |
+ e + ['--stderr'], stderr=STDOUT, universal_newlines=un) |
+ self._check_res(res, None, None, 0) |
self._run_test(fn) |