Improve testing by comparing behavior of subprocess to subprocess2

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
 import sys
 import time
 import unittest
 
 try:
   import fcntl
 except ImportError:
   fcntl = None
 
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
+import subprocess
 import subprocess2
 
 from testing_support import auto_stub
 
 # Method could be a function
 # pylint: disable=R0201
 
 
 def convert_to_crlf(string):
   """Unconditionally convert LF to CRLF."""
@@ skipping 43 matching lines @@
 
   def _fake_subprocess_Popen(self):
     """Mocks the base class subprocess.Popen only."""
     results = {}
     def __init__(self, args, **kwargs):
       assert not results
       results.update(kwargs)
       results['args'] = args
     def communicate():
       return None, None
-    self.mock(subprocess2.subprocess.Popen, '__init__', __init__)
-    self.mock(subprocess2.subprocess.Popen, 'communicate', communicate)
+    self.mock(subprocess.Popen, '__init__', __init__)
+    self.mock(subprocess.Popen, 'communicate', communicate)
     return results
 
   def test_check_call_defaults(self):
     results = self._fake_communicate()
     self.assertEquals(
         ('stdout', 'stderr'), subprocess2.check_call_out(['foo'], a=True))
     expected = {
         'args': ['foo'],
         'a':True,
     }
@@ skipping 54 matching lines @@
     self.assertEquals('stdout', subprocess2.check_output(['foo'], a=True))
     expected = {
         'args': ['foo'],
         'a':True,
         'stdin': subprocess2.VOID,
         'stdout': subprocess2.PIPE,
     }
     self.assertEquals(expected, results)
 
 
-class S2Test(unittest.TestCase):
+class BaseTestCase(unittest.TestCase):
   def setUp(self):
-    super(S2Test, self).setUp()
+    super(BaseTestCase, self).setUp()
     self.exe_path = __file__
     self.exe = [sys.executable, self.exe_path, '--child']
     self.states = {}
     if fcntl:
       for v in (sys.stdin, sys.stdout, sys.stderr):
         fileno = v.fileno()
         self.states[fileno] = fcntl.fcntl(fileno, fcntl.F_GETFL)
 
   def tearDown(self):
     for fileno, fl in self.states.iteritems():
       self.assertEquals(fl, fcntl.fcntl(fileno, fcntl.F_GETFL))
-    super(S2Test, self).tearDown()
+    super(BaseTestCase, self).tearDown()
 
+
+class RegressionTest(BaseTestCase):
+  # Regression tests to ensure that subprocess and subprocess2 have the same
+  # behavior.
+  def _run_test(self, function):
+    """Runs tests in 12 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
+
+    Once with subprocess, once with subprocess2.
+
+    First |function| argument is the convertion for the origianl expected LF

[Dirk Pranke, 2011/11/28 22:31:50]

typos: "conversion" and "original".

+    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
+    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):
+    """On exception, look if the exception members are set correctly."""
+    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()
+
+  def test_check_output_no_stdout(self):
+    for subp in (subprocess, subprocess2):
+      try:
+        subp.check_output(self.exe, stdout=subp.PIPE)
+        self.fail()
+      except ValueError:
+        pass
+
+  def test_check_output_throw_stdout(self):
+    def fn(c, e, un, subp):
+      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._run_test(fn)
+
+  def test_check_output_throw_no_stderr(self):
+    def fn(c, e, un, subp):
+      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._run_test(fn)
+
+  def test_check_output_throw_stderr(self):
+    def fn(c, e, un, subp):
+      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._run_test(fn)
+
+  def test_check_output_throw_stderr_stdout(self):
+    def fn(c, e, un, subp):
+      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._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)
+
+
+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 test_timeout(self):
-    out, returncode = subprocess2.communicate(
-        self.exe + ['--sleep_first', '--stdout'],
-        timeout=0.01,
-        stdout=subprocess2.PIPE,
-        shell=False)
-    self.assertEquals(subprocess2.TIMED_OUT, returncode)
-    self.assertEquals(('', None), out)
-
-  def test_check_output_no_stdout(self):
-    try:
-      subprocess2.check_output(self.exe, stdout=subprocess2.PIPE)
-      self.fail()
-    except TypeError:
-      pass
+    # timeout doesn't exist in subprocess.
+    def fn(c, e, un):
+      out, returncode = subprocess2.communicate(
+          self.exe + ['--sleep_first', '--stdout'],
+          timeout=0.01,
+          stdout=subprocess2.PIPE,
+          shell=False)
+      self.assertEquals(subprocess2.TIMED_OUT, returncode)
+      self.assertEquals(('', None), out)
+    self._run_test(fn)
 
   def test_stdout_void(self):
     def fn(c, e, un):
       (out, err), code = subprocess2.communicate(
           e + ['--stdout', '--stderr'],
           stdout=subprocess2.VOID,
           stderr=subprocess2.PIPE,
           universal_newlines=un)
       self.assertEquals(None, out)
       self.assertEquals(c('a\nbb\nccc\n'), err)
@@ skipping 30 matching lines @@
       (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)
     self._run_test(fn)
 
-  def test_check_output_throw_stdout(self):
-    def fn(c, e, un):
-      try:
-        subprocess2.check_output(
-            e + ['--fail', '--stdout'], universal_newlines=un)
-        self.fail()
-      except subprocess2.CalledProcessError, e:
-        self.assertEquals(c('A\nBB\nCCC\n'), e.stdout)
-        self.assertEquals(None, e.stderr)
-        self.assertEquals(64, e.returncode)
-    self._run_test(fn)
-
-  def test_check_output_throw_no_stderr(self):
-    def fn(c, e, un):
-      try:
-        subprocess2.check_output(
-            e + ['--fail', '--stderr'], universal_newlines=un)
-        self.fail()
-      except subprocess2.CalledProcessError, e:
-        self.assertEquals(c(''), e.stdout)
-        self.assertEquals(None, e.stderr)
-        self.assertEquals(64, e.returncode)
-    self._run_test(fn)
-
-  def test_check_output_throw_stderr(self):
-    def fn(c, e, un):
-      try:
-        subprocess2.check_output(
-            e + ['--fail', '--stderr'], stderr=subprocess2.PIPE,
-            universal_newlines=un)
-        self.fail()
-      except subprocess2.CalledProcessError, e:
-        self.assertEquals('', e.stdout)
-        self.assertEquals(c('a\nbb\nccc\n'), e.stderr)
-        self.assertEquals(64, e.returncode)
-    self._run_test(fn)
-
-  def test_check_output_throw_stderr_stdout(self):
-    def fn(c, e, un):
-      try:
-        subprocess2.check_output(
-            e + ['--fail', '--stderr'], stderr=subprocess2.STDOUT,
-            universal_newlines=un)
-        self.fail()
-      except subprocess2.CalledProcessError, e:
-        self.assertEquals(c('a\nbb\nccc\n'), e.stdout)
-        self.assertEquals(None, e.stderr)
-        self.assertEquals(64, e.returncode)
-    self._run_test(fn)
-
-  def test_check_call_throw(self):
-    try:
-      subprocess2.check_call(self.exe + ['--fail', '--stderr'])
-      self.fail()
-    except subprocess2.CalledProcessError, e:
-      self.assertEquals(None, e.stdout)
-      self.assertEquals(None, e.stderr)
-      self.assertEquals(64, e.returncode)
-
 
 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)
 
   parser = optparse.OptionParser()
@@ skipping 45 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()