Add subprocess2 unit 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 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
+
+
+class Subprocess2Test(unittest.TestCase):
+  # Can be mocked in a test.
+  TO_SAVE = ['Popen', 'call', 'check_call', 'capture', 'check_output']
+
+  def setUp(self):
+    self.exe_path = __file__
+    self.exe = [self.exe_path, '--child']
+    self.saved = dict(
+        (name, getattr(subprocess2, name)) for name in self.TO_SAVE)
+
+  def tearDown(self):
+    for name, value in self.saved.iteritems():
+      setattr(subprocess2, name, value)
+
+  @staticmethod
+  def _prep():
+    results = {}
+    def fake_call(args, **kwargs):
+      results.update(kwargs)
+      results['args'] = args
+      return ['stdout', 'stderr'], 0
+    subprocess2.call = fake_call
+    return results
+
+  def test_check_call_defaults(self):
+    results = self._prep()
+    self.assertEquals(
+        ['stdout', 'stderr'], subprocess2.check_call(['foo'], a=True))
+    expected = {
+        'args': ['foo'],
+        'a':True,
+    }
+    self.assertEquals(expected, results)
+
+  def test_check_output_defaults(self):
+    results = self._prep()
+    # It's discarding 'stderr' because it assumes stderr=subprocess2.STDOUT but
+    # fake_call() doesn't 'implement' that.
+    self.assertEquals('stdout', subprocess2.check_output(['foo'], a=True))
+    expected = {
+        'args': ['foo'],
+        'a':True,
+        'stdout': subprocess2.PIPE,
+        'stderr': subprocess2.STDOUT,
+    }
+    self.assertEquals(expected, results)
+
+  def test_timeout(self):
+    # It'd be better to not discard stdout.
+    out, returncode = subprocess2.call(
+        self.exe + ['--sleep', '--stdout'],
+        timeout=0.01,
+        stdout=subprocess2.PIPE)
+    self.assertEquals(-9, returncode)
+    self.assertEquals(['', None], out)
+
+  def test_void(self):
+    out = subprocess2.check_output(
+         self.exe + ['--stdout', '--stderr'],
+         stdout=subprocess2.VOID)
+    self.assertEquals(None, out)
+    out = subprocess2.check_output(
+         self.exe + ['--stdout', '--stderr'],
+         stderr=subprocess2.VOID)
+    self.assertEquals('A\nBB\nCCC\n', out)
+
+  def test_check_output_throw(self):
+    try:
+      subprocess2.check_output(self.exe + ['--fail', '--stderr'])
+      self.fail()
+    except subprocess2.CalledProcessError, e:
+      self.assertEquals('a\nbb\nccc\n', e.stdout)
+      self.assertEquals(None, e.stderr)
+      self.assertEquals(64, e.returncode)
+
+  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):
+  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')
+  options, args = parser.parse_args(args)
+  if args:
+    parser.error('Internal error')
+
+  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)
+  return options.return_value
+
+
+if __name__ == '__main__':
+  if len(sys.argv) > 1 and sys.argv[1] == '--child':
+    sys.exit(child_main(sys.argv[2:]))
+  unittest.main()