From a8e8163a06b9b4816ef37ae3c2f9649eff9baa95 Mon Sep 17 00:00:00 2001 From: "maruel@chromium.org" Date: Thu, 1 Dec 2011 00:35:24 +0000 Subject: [PATCH] Add Popen.shell, add more subprocess2 tests and make it more compact. R=dpranke@chromium.org BUG= TEST= Review URL: http://codereview.chromium.org/8750003 git-svn-id: svn://svn.chromium.org/chrome/trunk/tools/depot_tools@112351 0039d316-1c4b-4281-b951-d872f2087c98 --- subprocess2.py | 7 +++ tests/subprocess2_test.py | 101 ++++++++++++++++++++++---------------- 2 files changed, 67 insertions(+), 41 deletions(-) diff --git a/subprocess2.py b/subprocess2.py index 99eebcc1ae..c2f393a990 100644 --- a/subprocess2.py +++ b/subprocess2.py @@ -17,6 +17,7 @@ import tempfile import time import threading + # Constants forwarded from subprocess. PIPE = subprocess.PIPE STDOUT = subprocess.STDOUT @@ -182,8 +183,10 @@ class Popen(subprocess.Popen): fix('stderr') self.start = time.time() + self.shell = kwargs.get('shell', None) # Silence pylint on MacOSX self.returncode = None + try: super(Popen, self).__init__(args, **kwargs) except OSError, e: @@ -238,6 +241,10 @@ def communicate(args, timeout=None, **kwargs): with tempfile.TemporaryFile() as buff: kwargs['stdout'] = buff proc = Popen(args, **kwargs) + if proc.shell: + raise TypeError( + 'Using timeout and shell simultaneously will cause a process leak ' + 'since the shell will be killed instead of the child process.') if stdin is not None: proc.stdin.write(stdin) while proc.returncode is None: diff --git a/tests/subprocess2_test.py b/tests/subprocess2_test.py index 10290592d7..55936e5582 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)