| #!/usr/bin/env vpython3 |
| # 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 os |
| import sys |
| import unittest |
| |
| if sys.version_info.major == 2: |
| import mock |
| else: |
| from unittest import mock |
| |
| DEPOT_TOOLS = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) |
| sys.path.insert(0, DEPOT_TOOLS) |
| |
| import subprocess |
| import subprocess2 |
| |
| TEST_FILENAME = 'subprocess2_test_script.py' |
| |
| TEST_COMMAND = [ |
| sys.executable, |
| os.path.join(DEPOT_TOOLS, 'testing_support', TEST_FILENAME), |
| ] |
| |
| |
| class DefaultsTest(unittest.TestCase): |
| @mock.patch('subprocess2.communicate') |
| def test_check_call_defaults(self, mockCommunicate): |
| mockCommunicate.return_value = (('stdout', 'stderr'), 0) |
| self.assertEqual( |
| ('stdout', 'stderr'), subprocess2.check_call_out(['foo'], a=True)) |
| mockCommunicate.assert_called_with(['foo'], a=True) |
| |
| @mock.patch('subprocess2.communicate') |
| def test_capture_defaults(self, mockCommunicate): |
| mockCommunicate.return_value = (('stdout', 'stderr'), 0) |
| self.assertEqual( |
| 'stdout', subprocess2.capture(['foo'], a=True)) |
| mockCommunicate.assert_called_with( |
| ['foo'], a=True, stdin=subprocess2.VOID_INPUT, stdout=subprocess2.PIPE) |
| |
| @mock.patch('subprocess2.Popen') |
| def test_communicate_defaults(self, mockPopen): |
| mockPopen().communicate.return_value = ('bar', 'baz') |
| mockPopen().returncode = -8 |
| self.assertEqual( |
| (('bar', 'baz'), -8), subprocess2.communicate(['foo'], a=True)) |
| mockPopen.assert_called_with(['foo'], a=True) |
| |
| @mock.patch('os.environ', {}) |
| @mock.patch('subprocess.Popen.__init__') |
| def test_Popen_defaults(self, mockPopen): |
| with mock.patch('sys.platform', 'win32'): |
| subprocess2.Popen(['foo'], a=True) |
| mockPopen.assert_called_with(['foo'], a=True, shell=True) |
| |
| with mock.patch('sys.platform', 'non-win32'): |
| subprocess2.Popen(['foo'], a=True) |
| mockPopen.assert_called_with(['foo'], a=True, shell=False) |
| |
| def test_get_english_env(self): |
| with mock.patch('sys.platform', 'win32'): |
| self.assertIsNone(subprocess2.get_english_env({})) |
| |
| with mock.patch('sys.platform', 'non-win32'): |
| self.assertIsNone(subprocess2.get_english_env({})) |
| self.assertIsNone( |
| subprocess2.get_english_env({'LANG': 'en_XX', 'LANGUAGE': 'en_YY'})) |
| self.assertEqual( |
| {'LANG': 'en_US.UTF-8', 'LANGUAGE': 'en_US.UTF-8'}, |
| subprocess2.get_english_env({'LANG': 'bar', 'LANGUAGE': 'baz'})) |
| |
| @mock.patch('subprocess2.communicate') |
| def test_check_output_defaults(self, mockCommunicate): |
| mockCommunicate.return_value = (('stdout', 'stderr'), 0) |
| self.assertEqual('stdout', subprocess2.check_output(['foo'], a=True)) |
| mockCommunicate.assert_called_with( |
| ['foo'], a=True, stdin=subprocess2.VOID_INPUT, stdout=subprocess2.PIPE) |
| |
| @mock.patch('subprocess.Popen.__init__') |
| def test_env_type(self, mockPopen): |
| if sys.version_info.major != 2: |
| subprocess2.Popen(['foo'], env={b'key': b'value'}) |
| mockPopen.assert_called_with(['foo'], |
| env={'key': 'value'}, |
| shell=mock.ANY) |
| |
| |
| def _run_test(with_subprocess=True): |
| """Runs a tests in 12 combinations: |
| - With universal_newlines=True and False. |
| - With LF, CR, and CRLF output. |
| - With subprocess and subprocess2. |
| """ |
| subps = (subprocess2, subprocess) if with_subprocess else (subprocess2,) |
| no_op = lambda s: s |
| to_bytes = lambda s: s.encode() |
| to_cr_bytes = lambda s: s.replace('\n', '\r').encode() |
| to_crlf_bytes = lambda s: s.replace('\n', '\r\n').encode() |
| def wrapper(test): |
| def inner(self): |
| for subp in subps: |
| # universal_newlines = False |
| test(self, to_bytes, TEST_COMMAND, False, subp) |
| test(self, to_cr_bytes, TEST_COMMAND + ['--cr'], False, subp) |
| test(self, to_crlf_bytes, TEST_COMMAND + ['--crlf'], False, subp) |
| # universal_newlines = True |
| test(self, no_op, TEST_COMMAND, True, subp) |
| test(self, no_op, TEST_COMMAND + ['--cr'], True, subp) |
| test(self, no_op, TEST_COMMAND + ['--crlf'], True, subp) |
| |
| return inner |
| return wrapper |
| |
| |
| class SmokeTests(unittest.TestCase): |
| # Regression tests to ensure that subprocess and subprocess2 have the same |
| # behavior. |
| def _check_res(self, res, stdout, stderr, returncode): |
| (out, err), code = res |
| self.assertEqual(stdout, out) |
| self.assertEqual(stderr, err) |
| self.assertEqual(returncode, code) |
| |
| def _check_exception(self, subp, e, stdout, stderr, returncode): |
| """On exception, look if the exception members are set correctly.""" |
| self.assertEqual(returncode, e.returncode) |
| if subp is subprocess2 or sys.version_info.major == 3: |
| self.assertEqual(stdout, e.stdout) |
| self.assertEqual(stderr, e.stderr) |
| else: |
| # subprocess never save the output. |
| self.assertFalse(hasattr(e, 'stdout')) |
| self.assertFalse(hasattr(e, 'stderr')) |
| |
| def test_check_output_no_stdout(self): |
| for subp in (subprocess, subprocess2): |
| with self.assertRaises(ValueError): |
| subp.check_output(TEST_COMMAND, stdout=subp.PIPE) |
| |
| def test_print_exception(self): |
| with self.assertRaises(subprocess2.CalledProcessError) as e: |
| subprocess2.check_output(TEST_COMMAND + ['--fail', '--stdout']) |
| exception_str = str(e.exception) |
| # Windows escapes backslashes so check only filename |
| self.assertIn(TEST_FILENAME + ' --fail --stdout', exception_str) |
| self.assertIn(str(e.exception.returncode), exception_str) |
| self.assertIn(e.exception.stdout.decode('utf-8', 'ignore'), exception_str) |
| |
| @_run_test() |
| def test_check_output_throw_stdout(self, c, cmd, un, subp): |
| with self.assertRaises(subp.CalledProcessError) as e: |
| subp.check_output( |
| cmd + ['--fail', '--stdout'], universal_newlines=un) |
| self._check_exception(subp, e.exception, c('A\nBB\nCCC\n'), None, 64) |
| |
| @_run_test() |
| def test_check_output_throw_no_stderr(self, c, cmd, un, subp): |
| with self.assertRaises(subp.CalledProcessError) as e: |
| subp.check_output( |
| cmd + ['--fail', '--stderr'], universal_newlines=un) |
| self._check_exception(subp, e.exception, c(''), None, 64) |
| |
| @_run_test() |
| def test_check_output_throw_stderr(self, c, cmd, un, subp): |
| with self.assertRaises(subp.CalledProcessError) as e: |
| subp.check_output( |
| cmd + ['--fail', '--stderr'], |
| stderr=subp.PIPE, |
| universal_newlines=un) |
| self._check_exception(subp, e.exception, c(''), c('a\nbb\nccc\n'), 64) |
| |
| @_run_test() |
| def test_check_output_throw_stderr_stdout(self, c, cmd, un, subp): |
| with self.assertRaises(subp.CalledProcessError) as e: |
| subp.check_output( |
| cmd + ['--fail', '--stderr'], |
| stderr=subp.STDOUT, |
| universal_newlines=un) |
| self._check_exception(subp, e.exception, c('a\nbb\nccc\n'), None, 64) |
| |
| def test_check_call_throw(self): |
| for subp in (subprocess, subprocess2): |
| with self.assertRaises(subp.CalledProcessError) as e: |
| subp.check_call(TEST_COMMAND + ['--fail', '--stderr']) |
| self._check_exception(subp, e.exception, None, None, 64) |
| |
| @_run_test() |
| def test_redirect_stderr_to_stdout_pipe(self, c, cmd, un, subp): |
| # stderr output into stdout. |
| proc = subp.Popen( |
| cmd + ['--stderr'], |
| stdout=subp.PIPE, |
| stderr=subp.STDOUT, |
| universal_newlines=un) |
| res = proc.communicate(), proc.returncode |
| self._check_res(res, c('a\nbb\nccc\n'), None, 0) |
| |
| @_run_test() |
| def test_redirect_stderr_to_stdout(self, c, cmd, un, subp): |
| # stderr output into stdout but stdout is not piped. |
| proc = subp.Popen( |
| cmd + ['--stderr'], stderr=subprocess2.STDOUT, universal_newlines=un) |
| res = proc.communicate(), proc.returncode |
| self._check_res(res, None, None, 0) |
| |
| @_run_test() |
| def test_stderr(self, c, cmd, un, subp): |
| cmd = ['expr', '1', '/', '0'] |
| if sys.platform == 'win32': |
| cmd = ['cmd.exe', '/c', 'exit', '1'] |
| p1 = subprocess.Popen(cmd, stderr=subprocess.PIPE, shell=False) |
| p2 = subprocess2.Popen(cmd, stderr=subprocess.PIPE, shell=False) |
| r1 = p1.communicate() |
| r2 = p2.communicate() |
| self.assertEqual(r1, r2) |
| |
| @_run_test(with_subprocess=False) |
| def test_stdin(self, c, cmd, un, subp): |
| stdin = c('0123456789') |
| res = subprocess2.communicate( |
| cmd + ['--read'], |
| stdin=stdin, |
| universal_newlines=un) |
| self._check_res(res, None, None, 10) |
| |
| @_run_test(with_subprocess=False) |
| def test_stdin_empty(self, c, cmd, un, subp): |
| stdin = c('') |
| res = subprocess2.communicate( |
| cmd + ['--read'], |
| stdin=stdin, |
| universal_newlines=un) |
| self._check_res(res, None, None, 0) |
| |
| def test_stdin_void(self): |
| res = subprocess2.communicate( |
| TEST_COMMAND + ['--read'], |
| stdin=subprocess2.VOID_INPUT) |
| self._check_res(res, None, None, 0) |
| |
| @_run_test(with_subprocess=False) |
| def test_stdin_void_stdout(self, c, cmd, un, subp): |
| # Make sure a mix ofsubprocess2.VOID andsubprocess2.PIPE works. |
| res = subprocess2.communicate( |
| cmd + ['--stdout', '--read'], |
| stdin=subprocess2.VOID_INPUT, |
| stdout=subprocess2.PIPE, |
| universal_newlines=un, |
| shell=False) |
| self._check_res(res, c('A\nBB\nCCC\n'), None, 0) |
| |
| @_run_test(with_subprocess=False) |
| def test_stdout_void(self, c, cmd, un, subp): |
| res = subprocess2.communicate( |
| cmd + ['--stdout', '--stderr'], |
| stdout=subprocess2.VOID, |
| stderr=subprocess2.PIPE, |
| universal_newlines=un) |
| self._check_res(res, None, c('a\nbb\nccc\n'), 0) |
| |
| @_run_test(with_subprocess=False) |
| def test_stderr_void(self, c, cmd, un, subp): |
| res = subprocess2.communicate( |
| cmd + ['--stdout', '--stderr'], |
| stdout=subprocess2.PIPE, |
| stderr=subprocess2.VOID, |
| universal_newlines=un) |
| self._check_res(res, c('A\nBB\nCCC\n'), None, 0) |
| |
| @_run_test(with_subprocess=False) |
| def test_stdout_void_stderr_redirect(self, c, cmd, un, subp): |
| res = subprocess2.communicate( |
| cmd + ['--stdout', '--stderr'], |
| stdout=subprocess2.VOID, |
| stderr=subprocess2.STDOUT, |
| universal_newlines=un) |
| self._check_res(res, None, None, 0) |
| |
| |
| if __name__ == '__main__': |
| unittest.main() |