ABANDONED. Use gclient_utils instead.

tests/subprocess2_test.py

 #!/usr/bin/env python
 # Copyright (c) 2012 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 cStringIO
 import logging
 import optparse
 import os
 import sys
 import time
 import unittest
 
 try:
   import fcntl  # pylint: disable=F0401
 except ImportError:
@@ skipping 140 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 BaseTestCase(unittest.TestCase):
+class BaseTestCase(auto_stub.TestCase):
   def setUp(self):
     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)
 
@@ skipping 50 matching lines @@
       self.assertEquals(stderr, e.stderr)
     else:
       self.fail()
 
   def test_check_output_no_stdout(self):
     try:
       subprocess2.check_output(self.exe, stdout=subprocess2.PIPE)
       self.fail()
     except ValueError:
       pass
-    
+
     if (sys.version_info[0] * 10 + sys.version_info[1]) >= 27:
       # python 2.7+
       try:
         # pylint: disable=E1101
         subprocess.check_output(self.exe, stdout=subprocess.PIPE)
         self.fail()
       except ValueError:
         pass
 
   def test_check_output_throw_stdout(self):
@@ skipping 360 matching lines @@
     proc = subprocess2.Popen(
         self.exe + ['--stdout', '--sleep_first'], stdout=PIPE)
     res = proc.communicate(nag_timer=3), proc.returncode
     l.removeFilter(f)
     self._check_res(res, 'A\nBB\nCCC\n', None, 0)
     expected = ['No output for 3 seconds from command:', proc.cmd_str,
                 'No output for 6 seconds from command:', proc.cmd_str,
                 'No output for 9 seconds from command:', proc.cmd_str]
     self.assertEquals(w, expected)
 
-    
+  def test_communicate_and_stream(self):
+    self.mock(sys, 'stdout', cStringIO.StringIO())
+    self.mock(sys, 'stderr', cStringIO.StringIO())
+    (stdout, stderr), returncode = subprocess2.communicate_and_stream(
+        self.exe + ['--stderr', '--stdout', '--fail'])
+    self.assertEquals(returncode, 64)
+    self.assertEquals('A\nBB\nCCC\n', sys.stdout.getvalue())
+    self.assertEquals('A\nBB\nCCC\n', stdout)
+    self.assertEquals('a\nbb\nccc\n', sys.stderr.getvalue())
+    self.assertEquals('a\nbb\nccc\n', stderr)
+
+  def test_communicate_and_stream_partial_pipe(self):
+    self.mock(sys, 'stdout', cStringIO.StringIO())
+    self.mock(sys, 'stderr', cStringIO.StringIO())
+    (stdout, stderr), returncode = subprocess2.communicate_and_stream(
+        self.exe + ['--stderr', '--stdout'], stdout=PIPE)
+    self.assertEquals(returncode, 0)
+    self.assertEquals('', sys.stdout.getvalue())
+    self.assertEquals('A\nBB\nCCC\n', stdout)
+    self.assertEquals('a\nbb\nccc\n', sys.stderr.getvalue())
+    self.assertEquals('a\nbb\nccc\n', stderr)
+
+  def test_communicate_and_stream_partial_void(self):
+    self.mock(sys, 'stdout', cStringIO.StringIO())
+    self.mock(sys, 'stderr', cStringIO.StringIO())
+    (stdout, stderr), returncode = subprocess2.communicate_and_stream(
+        self.exe + ['--stderr', '--stdout', '--fail'], stderr=VOID)
+    self.assertEquals(returncode, 64)
+    self.assertEquals('A\nBB\nCCC\n', sys.stdout.getvalue())
+    self.assertEquals('A\nBB\nCCC\n', stdout)
+    self.assertEquals('', sys.stderr.getvalue())
+    self.assertIsNone(stderr)
+
+  def test_communicate_and_stream_redirect(self):
+    self.mock(sys, 'stdout', cStringIO.StringIO())
+    self.mock(sys, 'stderr', cStringIO.StringIO())
+    (stdout, stderr), returncode = subprocess2.communicate_and_stream(
+        self.exe + ['--stderr', '--stdout', '--fail'], stderr=STDOUT)
+    self.assertEquals(returncode, 64)
+    self.assertEquals('', sys.stderr.getvalue())
+    self.assertIsNone(stderr)
+    self.assertEquals(sys.stdout.getvalue(), stdout)
+    # Because stderr/stdout is interwoven arbitrarily,
+    # just check that no output is missed, but not the order.
+    self.assertEquals(sorted(stdout), sorted('A\nBB\nCCC\n' + 'a\nbb\nccc\n'))
+
+
+
 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()
   parser.add_option(
@@ 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()