Add Popen.shell, add more subprocess2 tests and make it more compact.

tests/subprocess2_test.py

 #!/usr/bin/env python
 # Copyright (c) 2011 The Chromium Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
 """Unit tests for subprocess2.py."""
 
 import logging
 import optparse
 import os
@@ skipping 10 matching lines @@
 
 import subprocess
 import subprocess2
 
 from testing_support import auto_stub
 
 # Method could be a function
 # 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')
 
 
 def convert_to_cr(string):
   """Unconditionally convert LF to CR."""
   return string.replace('\n', '\r')
 
 
@@ skipping 154 matching lines @@
     """
     noop = lambda x: x
     for subp in (subprocess, subprocess2):
       function(noop, self.exe, False, subp)
       function(convert_to_cr, self.exe + ['--cr'], False, subp)
       function(convert_to_crlf, self.exe + ['--crlf'], False, subp)
       function(noop, self.exe, True, subp)
       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'))
       self.assertFalse(hasattr(e, 'stderr'))
     elif subp is subprocess2:
       self.assertEquals(stdout, e.stdout)
       self.assertEquals(stderr, e.stderr)
     else:
       self.fail()
 
@@ skipping 15 matching lines @@
 
   def test_check_output_throw_stdout(self):
     def fn(c, e, un, subp):
       if not hasattr(subp, 'check_output'):
         return
       try:
         subp.check_output(
             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):
     def fn(c, e, un, subp):
       if not hasattr(subp, 'check_output'):
         return
       try:
         subp.check_output(
             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):
     def fn(c, e, un, subp):
       if not hasattr(subp, 'check_output'):
         return
       try:
         subp.check_output(
             e + ['--fail', '--stderr'],
             stderr=subp.PIPE,
             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):
     def fn(c, e, un, subp):
       if not hasattr(subp, 'check_output'):
         return
       try:
         subp.check_output(
             e + ['--fail', '--stderr'],
             stderr=subp.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_call_throw(self):
     for subp in (subprocess, subprocess2):
       try:
         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):
   # Tests that can only run in subprocess2, e.g. new functionalities.
   # In particular, subprocess2.communicate() doesn't exist in subprocess.
   def _run_test(self, function):
     """Runs tests in 6 combinations:
     - LF output with universal_newlines=False
     - CR output with universal_newlines=False
     - CRLF output with universal_newlines=False
     - LF output with universal_newlines=True
     - CR output with universal_newlines=True
     - CRLF output with universal_newlines=True
 
     First |function| argument is the convertion for the origianl expected LF
     string to the right EOL.
     Second |function| argument is the executable and initial flag to run, to
     control what EOL is used by the child process.
     Third |function| argument is universal_newlines value.
     """
     noop = lambda x: x
     function(noop, self.exe, False)
     function(convert_to_cr, self.exe + ['--cr'], False)
     function(convert_to_crlf, self.exe + ['--crlf'], False)
     function(noop, self.exe, True)
     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)
 
 
 def child_main(args):
   if sys.platform == 'win32':
     # Annoying, make sure the output is not translated on Windows.
     # pylint: disable=E1101,F0401
     import msvcrt
     msvcrt.setmode(sys.stdout.fileno(), os.O_BINARY)
     msvcrt.setmode(sys.stderr.fileno(), os.O_BINARY)
@@ skipping 47 matching lines @@
   return options.return_value
 
 
 if __name__ == '__main__':
   logging.basicConfig(level=
       [logging.WARNING, logging.INFO, logging.DEBUG][
         min(2, sys.argv.count('-v'))])
   if len(sys.argv) > 1 and sys.argv[1] == '--child':
     sys.exit(child_main(sys.argv[2:]))
   unittest.main()