Add callback support for stdout and stderr.

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
 
 ROOT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 sys.path.insert(0, ROOT_DIR)
 
 import subprocess2
 
 # Method could be a function
 # pylint: disable=R0201
 
+
+def convert(string):
+  """Converts string to CRLF on Windows."""
+  if sys.platform == 'win32':
+    return string.replace('\n', '\r\n')
+  return string
+
+
 class Subprocess2Test(unittest.TestCase):
   # Can be mocked in a test.
   TO_SAVE = {
     subprocess2: [
       'Popen', 'communicate', 'call', 'check_call', 'capture', 'check_output'],
     subprocess2.subprocess: ['Popen'],
   }
 
   def setUp(self):
     self.exe_path = __file__
@@ skipping 108 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)
 
   def test_timeout(self):
-    # It'd be better to not discard stdout.
     out, returncode = subprocess2.communicate(
         self.exe + ['--sleep', '--stdout'],
         timeout=0.01,
-        stdout=subprocess2.PIPE)
+        stdout=subprocess2.PIPE,
+        shell=False)
     self.assertEquals(subprocess2.TIMED_OUT, returncode)
-    self.assertEquals(['', None], out)
+    self.assertEquals(('', None), out)
+
+  def test_timeout_shell_throws(self):
+    try:
+      subprocess2.communicate(
+          self.exe + ['--sleep', '--stdout'],
+          timeout=0.01,
+          stdout=subprocess2.PIPE,
+          shell=True)
+      self.fail()
+    except TypeError:
+      pass
 
   def test_check_output_no_stdout(self):
     try:
       subprocess2.check_output(self.exe, stdout=subprocess2.PIPE)
       self.fail()
     except TypeError:
       pass
 
   def test_stdout_void(self):
     (out, err), code = subprocess2.communicate(
          self.exe + ['--stdout', '--stderr'],
          stdout=subprocess2.VOID,
          stderr=subprocess2.PIPE)
     self.assertEquals(None, out)
-    expected = 'a\nbb\nccc\n'
-    if sys.platform == 'win32':
-      expected = expected.replace('\n', '\r\n')
-    self.assertEquals(expected, err)
+    self.assertEquals(convert('a\nbb\nccc\n'), err)
     self.assertEquals(0, code)
 
   def test_stderr_void(self):
     (out, err), code = subprocess2.communicate(
          self.exe + ['--stdout', '--stderr'],
          universal_newlines=True,
          stdout=subprocess2.PIPE,
          stderr=subprocess2.VOID)
     self.assertEquals('A\nBB\nCCC\n', out)
     self.assertEquals(None, err)
@@ skipping 43 matching lines @@
 
   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 test_check_output_tee_stderr(self):
+    stderr = []
+    out, returncode = subprocess2.communicate(
+        self.exe + ['--stderr'], stderr=stderr.append)
+    self.assertEquals(convert('a\nbb\nccc\n'), ''.join(stderr))
+    self.assertEquals((None, None), out)
+    self.assertEquals(0, returncode)
+
+  def test_check_output_tee_stdout_stderr(self):
+    stdout = []
+    stderr = []
+    out, returncode = subprocess2.communicate(
+        self.exe + ['--stdout', '--stderr'],
+        stdout=stdout.append,
+        stderr=stderr.append)
+    self.assertEquals(convert('A\nBB\nCCC\n'), ''.join(stdout))
+    self.assertEquals(convert('a\nbb\nccc\n'), ''.join(stderr))
+    self.assertEquals((None, None), out)
+    self.assertEquals(0, returncode)
+
+  def test_check_output_tee_stdin(self):
+    stdout = []
+    stdin = '0123456789'
+    out, returncode = subprocess2.communicate(
+        self.exe + ['--stdout', '--read'], stdin=stdin, stdout=stdout.append)
+    self.assertEquals(convert('A\nBB\nCCC\n'), ''.join(stdout))
+    self.assertEquals((None, None), out)
+    self.assertEquals(0, returncode)
+
+  def test_check_output_tee_throw(self):
+    stderr = []
+    try:
+      subprocess2.check_output(
+          self.exe + ['--stderr', '--fail'], stderr=stderr.append)
+      self.fail()
+    except subprocess2.CalledProcessError, e:
+      self.assertEquals(convert('a\nbb\nccc\n'), ''.join(stderr))
+      self.assertEquals('', e.stdout)
+      self.assertEquals(None, e.stderr)
+      self.assertEquals(64, e.returncode)
+
+  def test_check_output_tee_large(self):
+    stdout = []
+    # Read 128kb. On my workstation it takes >2s. Welcome to 2011.
+    out, returncode = subprocess2.communicate(
+        self.exe + ['--large'], stdout=stdout.append)
+    self.assertEquals(128*1024, len(''.join(stdout)))
+    self.assertEquals((None, None), out)
+    self.assertEquals(0, returncode)
+
+  def test_check_output_tee_large_stdin(self):
+    stdout = []
+    # Write 128kb.
+    stdin = '0123456789abcdef' * (8*1024)
+    out, returncode = subprocess2.communicate(
+        self.exe + ['--large', '--read'], stdin=stdin, stdout=stdout.append)
+    self.assertEquals(128*1024, len(''.join(stdout)))
+    self.assertEquals((None, None), out)
+    self.assertEquals(0, returncode)
+
 
 def child_main(args):
   parser = optparse.OptionParser()
   parser.add_option(
       '--fail',
       dest='return_value',
       action='store_const',
       default=0,
       const=64)
   parser.add_option('--stdout', action='store_true')
   parser.add_option('--stderr', action='store_true')
   parser.add_option('--sleep', action='store_true')
+  parser.add_option('--large', action='store_true')
+  parser.add_option('--read', action='store_true')
   options, args = parser.parse_args(args)
   if args:
     parser.error('Internal error')
+  if options.sleep:
+    time.sleep(10)
 
   def do(string):
     if options.stdout:
       print >> sys.stdout, string.upper()
     if options.stderr:
       print >> sys.stderr, string.lower()
 
   do('A')
   do('BB')
   do('CCC')
-  if options.sleep:
-    time.sleep(10)
+  if options.large:
+    # Print 128kb.
+    string = '0123456789abcdef' * (8*1024)
+    sys.stdout.write(string)
+  if options.read:
+    try:
+      while sys.stdin.read():
+        pass
+    except OSError:
+      pass
   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()