Reimplement r109239 but using Popen.communicate() instead.

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 59 matching lines @@
   def _fake_Popen(self):
     """Mocks the whole subprocess2.Popen class."""
     results = {}
     class fake_Popen(object):
       returncode = -8
       def __init__(self, args, **kwargs):
         assert not results
         results.update(kwargs)
         results['args'] = args
       @staticmethod
-      def communicate():
+      def communicate(input=None, timeout=None):  # pylint: disable=W0622
         return None, None
     self.mock(subprocess2, 'Popen', fake_Popen)
     return results
 
   def _fake_subprocess_Popen(self):
     """Mocks the base class subprocess.Popen only."""
     results = {}
     def __init__(self, args, **kwargs):
       assert not results
       results.update(kwargs)
@@ skipping 82 matching lines @@
     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(BaseTestCase, self).tearDown()
 
+  def _check_res(self, res, stdout, stderr, returncode):
+    (out, err), code = res
+    self.assertEquals(stdout, out)
+    self.assertEquals(stderr, err)
+    self.assertEquals(returncode, code)
+
 
 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
@@ skipping 99 matching lines @@
     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_exception(subp, e, None, None, 64)
 
+  def test_redirect_stderr_to_stdout_pipe(self):
+    def fn(c, e, un, subp):
+      # stderr output into stdout.
+      proc = subp.Popen(
+          e + ['--stderr'],
+          stdout=subp.PIPE,
+          stderr=subp.STDOUT,
+          universal_newlines=un)
+      res = proc.communicate(), proc.returncode
+      self._check_res(res, c('a\nbb\nccc\n'), None, 0)
+    self._run_test(fn)
+
+  def test_redirect_stderr_to_stdout(self):
+    def fn(c, e, un, subp):
+      # stderr output into stdout but stdout is not piped.
+      proc = subp.Popen(
+          e + ['--stderr'], stderr=STDOUT, universal_newlines=un)
+      res = proc.communicate(), proc.returncode
+      self._check_res(res, None, None, 0)
+    self._run_test(fn)
+
 
 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 _check_exception(self, e, stdout, stderr, returncode):
+    """On exception, look if the exception members are set correctly."""
+    self.assertEquals(returncode, e.returncode)
+    self.assertEquals(stdout, e.stdout)
+    self.assertEquals(stderr, e.stderr)
 
   def test_timeout(self):
     # timeout doesn't exist in subprocess.
     def fn(c, e, un):
       res = subprocess2.communicate(
           self.exe + ['--sleep_first', '--stdout'],
           timeout=0.01,
           stdout=PIPE,
           shell=False)
       self._check_res(res, '', None, TIMED_OUT)
@@ skipping 32 matching lines @@
   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 test_tee_stderr(self):
     def fn(c, e, un):
-      # stderr output into stdout.
+      stderr = []
       res = subprocess2.communicate(
-          e + ['--stderr'],
-          stdout=PIPE,
-          stderr=STDOUT,
-          universal_newlines=un)
-      self._check_res(res, c('a\nbb\nccc\n'), None, 0)
+          e + ['--stderr'], stderr=stderr.append, universal_newlines=un)
+      self.assertEquals(c('a\nbb\nccc\n'), ''.join(stderr))
+      self._check_res(res, None, None, 0)
     self._run_test(fn)
 
-  def test_check_output_redirect_stderr_to_stdout(self):
+  def test_tee_stdout_stderr(self):
     def fn(c, e, un):
-      # stderr output into stdout but stdout is not piped.
+      stdout = []
+      stderr = []
       res = subprocess2.communicate(
-          e + ['--stderr'], stderr=STDOUT, universal_newlines=un)
+          e + ['--stdout', '--stderr'],
+          stdout=stdout.append,
+          stderr=stderr.append,
+          universal_newlines=un)
+      self.assertEquals(c('A\nBB\nCCC\n'), ''.join(stdout))
+      self.assertEquals(c('a\nbb\nccc\n'), ''.join(stderr))
       self._check_res(res, None, None, 0)
     self._run_test(fn)
 
+  def test_tee_stdin(self):
+    def fn(c, e, un):
+      stdout = []
+      stdin = '0123456789'
+      res = subprocess2.communicate(
+          e + ['--stdout', '--read'], stdin=stdin, stdout=stdout.append,
+          universal_newlines=un)
+      self.assertEquals(c('A\nBB\nCCC\n'), ''.join(stdout))
+      self._check_res(res, None, None, 0)
+    self._run_test(fn)
+
+  def test_tee_throw(self):
+    def fn(c, e, un):
+      stderr = []
+      try:
+        subprocess2.check_output(
+            e + ['--stderr', '--fail'], stderr=stderr.append,
+            universal_newlines=un)
+        self.fail()
+      except subprocess2.CalledProcessError, e:
+        self._check_exception(e, '', None, 64)
+        self.assertEquals(c('a\nbb\nccc\n'), ''.join(stderr))
+    self._run_test(fn)
+
+  def test_tee_timeout_stdout_void(self):
+    def fn(c, e, un):
+      stderr = []
+      res = subprocess2.communicate(
+          e + ['--stdout', '--stderr', '--fail'],
+          stdout=VOID,
+          stderr=stderr.append,
+          shell=False,
+          timeout=10,
+          universal_newlines=un)
+      self._check_res(res, None, None, 64)
+      self.assertEquals(c('a\nbb\nccc\n'), ''.join(stderr))
+    self._run_test(fn)
+
+  def test_tee_timeout_stderr_void(self):
+    def fn(c, e, un):
+      stdout = []
+      res = subprocess2.communicate(
+          e + ['--stdout', '--stderr', '--fail'],
+          stdout=stdout.append,
+          stderr=VOID,
+          shell=False,
+          timeout=10,
+          universal_newlines=un)
+      self._check_res(res, None, None, 64)
+      self.assertEquals(c('A\nBB\nCCC\n'), ''.join(stdout))
+    self._run_test(fn)
+
+  def test_tee_timeout_stderr_stdout(self):
+    def fn(c, e, un):
+      stdout = []
+      res = subprocess2.communicate(
+          e + ['--stdout', '--stderr', '--fail'],
+          stdout=stdout.append,
+          stderr=STDOUT,
+          shell=False,
+          timeout=10,
+          universal_newlines=un)
+      self._check_res(res, None, None, 64)
+      # Ordering is random due to buffering.
+      self.assertEquals(
+          set(c('a\nbb\nccc\nA\nBB\nCCC\n').splitlines(True)),
+          set(''.join(stdout).splitlines(True)))
+    self._run_test(fn)
+
+  def test_tee_large(self):
+    stdout = []
+    # Read 128kb. On my workstation it takes >2s. Welcome to 2011.
+    res = subprocess2.communicate(self.exe + ['--large'], stdout=stdout.append)
+    self.assertEquals(128*1024, len(''.join(stdout)))
+    self._check_res(res, None, None, 0)
+
+  def test_tee_large_stdin(self):
+    stdout = []
+    # Write 128kb.
+    stdin = '0123456789abcdef' * (8*1024)
+    res = subprocess2.communicate(
+        self.exe + ['--large', '--read'], stdin=stdin, stdout=stdout.append)
+    self.assertEquals(128*1024, len(''.join(stdout)))
+    self._check_res(res, None, None, 0)
+
+  def test_tee_cb_throw(self):
+    # Having a callback throwing up should not cause side-effects. It's a bit
+    # hard to measure.
+    class Blow(Exception):
+      pass
+    def blow(_):
+      raise Blow()
+    proc = subprocess2.Popen(self.exe + ['--stdout'], stdout=blow)
+    try:
+      proc.communicate()
+      self.fail()
+    except Blow:
+      self.assertNotEquals(0, proc.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()