Improve subprocess2 stdin tests.

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 362 matching lines @@
   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_stdin(self):
+    def fn(c, e, un):
+      stdin = '0123456789'
+      res = subprocess2.communicate(
+          e + ['--read'],
+          stdin=stdin,
+          universal_newlines=un)
+      self._check_res(res, None, None, 10)
+    self._run_test(fn)
+
+  def test_stdin_unicode(self):
+    def fn(c, e, un):
+      stdin = u'0123456789'
+      res = subprocess2.communicate(
+          e + ['--read'],
+          stdin=stdin,
+          universal_newlines=un)
+      self._check_res(res, None, None, 10)
+    self._run_test(fn)
+
+  def test_stdin_void(self):
+    res = subprocess2.communicate(self.exe + ['--read'], stdin=VOID)
+    self._check_res(res, None, None, 0)
+
+  def test_stdin_void_stdout_timeout(self):
+    # Make sure a mix of VOID, PIPE and timeout works.
+    def fn(c, e, un):
+      res = subprocess2.communicate(
+          e + ['--stdout', '--read'],
+          stdin=VOID,
+          stdout=PIPE,
+          timeout=10,
+          universal_newlines=un)
+      self._check_res(res, c('A\nBB\nCCC\n'), None, 0)
+    self._run_test(fn)
+
   def test_stdout_void(self):
     def fn(c, e, un):
       res = subprocess2.communicate(
           e + ['--stdout', '--stderr'],
           stdout=VOID,
           stderr=PIPE,
           universal_newlines=un)
       self._check_res(res, None, c('a\nbb\nccc\n'), 0)
     self._run_test(fn)
 
@@ skipping 35 matching lines @@
           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):
+      # Mix of stdin input and stdout callback.
       stdout = []
       stdin = '0123456789'
       res = subprocess2.communicate(
-          e + ['--stdout', '--read'], stdin=stdin, stdout=stdout.append,
+          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._check_res(res, None, None, 10)
     self._run_test(fn)
 
   def test_tee_throw(self):
     def fn(c, e, un):
+      # Make sure failure still returns stderr completely.
       stderr = []
       try:
         subprocess2.check_output(
-            e + ['--stderr', '--fail'], stderr=stderr.append,
+            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 = []
@@ skipping 110 matching lines @@
       sys.stderr.write(options.eol)
 
   do('A')
   do('BB')
   do('CCC')
   if options.large:
     # Print 128kb.
     string = '0123456789abcdef' * (8*1024)
     sys.stdout.write(string)
   if options.read:
+    assert options.return_value is 0
     try:
-      while sys.stdin.read():
-        pass
+      while sys.stdin.read(1):
+        options.return_value += 1
     except OSError:
       pass
   if options.sleep_last:
     time.sleep(10)
   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()