diff --git a/subprocess2.py b/subprocess2.py index c9057ac6a..f4a9a2f39 100644 --- a/subprocess2.py +++ b/subprocess2.py @@ -233,9 +233,11 @@ def capture(args, **kwargs): - Discards returncode. - Discards stderr. By default sets stderr=STDOUT. """ + if kwargs.get('stdout') is None: + kwargs['stdout'] = PIPE if kwargs.get('stderr') is None: kwargs['stderr'] = STDOUT - return call(args, stdout=PIPE, **kwargs)[0][0] + return call(args, **kwargs)[0][0] def check_output(args, **kwargs): @@ -247,6 +249,8 @@ def check_output(args, **kwargs): - Throws if return code is not 0. - Works even prior to python 2.7. """ + if kwargs.get('stdout') is None: + kwargs['stdout'] = PIPE if kwargs.get('stderr') is None: kwargs['stderr'] = STDOUT - return check_call(args, stdout=PIPE, **kwargs)[0] + return check_call(args, **kwargs)[0] diff --git a/tests/subprocess2_test.py b/tests/subprocess2_test.py new file mode 100755 index 000000000..4f289026f --- /dev/null +++ b/tests/subprocess2_test.py @@ -0,0 +1,137 @@ +#!/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()