1 #!/usr/bin/env python 2 # 3 # Copyright 2017 - The Android Open Source Project 4 # 5 # Licensed under the Apache License, Version 2.0 (the "License"); 6 # you may not use this file except in compliance with the License. 7 # You may obtain a copy of the License at 8 # 9 # http://www.apache.org/licenses/LICENSE-2.0 10 # 11 # Unless required by applicable law or agreed to in writing, software 12 # distributed under the License is distributed on an "AS IS" BASIS, 13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14 # See the License for the specific language governing permissions and 15 # limitations under the License. 16 17 """Unit test for gsi_util.utils.cmd_utils.""" 18 19 import logging 20 from logging import handlers 21 import shutil 22 import subprocess 23 import tempfile 24 import unittest 25 26 from gsi_util.utils import cmd_utils 27 28 29 class RunCommandTest(unittest.TestCase): 30 31 def setUp(self): 32 """Sets up logging output for assert checks.""" 33 log_entries = self._log_entries = [] 34 35 class Target(object): 36 """The target handler to store log output.""" 37 38 def handle(self, record): 39 log_entries.append((record.levelname, record.msg % record.args)) 40 41 self._handler = handlers.MemoryHandler(capacity=0, target=Target()) 42 logging.getLogger().addHandler(self._handler) 43 44 def tearDown(self): 45 """Removes logging handler.""" 46 logging.getLogger().removeHandler(self._handler) 47 48 def test_command_sequence(self): 49 result = cmd_utils.run_command(['echo', '123']) 50 self.assertEqual((0, None, None), result) 51 self.assertEqual(('INFO', "Executed command: 'echo 123' (ret: 0)"), 52 self._log_entries[0]) 53 54 def test_shell_command(self): 55 result = cmd_utils.run_command('echo uses shell', shell=True) 56 self.assertEqual((0, None, None), result) 57 self.assertEqual(('INFO', "Executed command: 'echo uses shell' (ret: 0)"), 58 self._log_entries[0]) 59 60 def test_log_stdout(self): 61 result = cmd_utils.run_command(['echo', '456'], raise_on_error=False, 62 log_stdout=True) 63 self.assertEqual((0, None, None), result) 64 self.assertEqual(('INFO', "Executed command: 'echo 456' (ret: 0)"), 65 self._log_entries[0]) 66 self.assertEqual(('INFO', " stdout: '456\\n'"), 67 self._log_entries[1]) 68 69 def test_log_stderr(self): 70 error_cmd = 'echo foo; echo bar; (echo 123; echo 456;)>&2; exit 3' 71 result = cmd_utils.run_command(error_cmd, shell=True, raise_on_error=False, 72 log_stderr=True) 73 self.assertEqual((3, None, None), result) 74 self.assertEqual( 75 ('ERROR', 'Executed command: %r (ret: %d)' % (error_cmd, 3)), 76 self._log_entries[0]) 77 self.assertEqual(('ERROR', " stderr: '123\\n456\\n'"), 78 self._log_entries[1]) 79 80 def test_log_stdout_and_log_stderr(self): 81 error_cmd = 'echo foo; echo bar; (echo 123; echo 456;)>&2; exit 5' 82 result = cmd_utils.run_command(error_cmd, shell=True, raise_on_error=False, 83 log_stdout=True, log_stderr=True) 84 self.assertEqual((5, None, None), result) 85 self.assertEqual( 86 ('ERROR', 'Executed command: %r (ret: %d)' % (error_cmd, 5)), 87 self._log_entries[0]) 88 self.assertEqual(('ERROR', " stdout: 'foo\\nbar\\n'"), 89 self._log_entries[1]) 90 self.assertEqual(('ERROR', " stderr: '123\\n456\\n'"), 91 self._log_entries[2]) 92 93 def test_read_stdout(self): 94 result = cmd_utils.run_command('echo 123; echo 456; exit 7', shell=True, 95 read_stdout=True, raise_on_error=False) 96 self.assertEqual(7, result.returncode) 97 self.assertEqual('123\n456\n', result.stdoutdata) 98 self.assertEqual(None, result.stderrdata) 99 100 def test_read_stderr(self): 101 result = cmd_utils.run_command('(echo error1; echo error2)>&2; exit 9', 102 shell=True, read_stderr=True, 103 raise_on_error=False) 104 self.assertEqual(9, result.returncode) 105 self.assertEqual(None, result.stdoutdata) 106 self.assertEqual('error1\nerror2\n', result.stderrdata) 107 108 def test_read_stdout_and_stderr(self): 109 result = cmd_utils.run_command('echo foo; echo bar>&2; exit 11', 110 shell=True, read_stdout=True, 111 read_stderr=True, raise_on_error=False) 112 self.assertEqual(11, result.returncode) 113 self.assertEqual('foo\n', result.stdoutdata) 114 self.assertEqual('bar\n', result.stderrdata) 115 116 def test_raise_on_error(self): 117 error_cmd = 'echo foo; exit 13' 118 with self.assertRaises(subprocess.CalledProcessError) as context_manager: 119 cmd_utils.run_command(error_cmd, shell=True, raise_on_error=True) 120 proc_err = context_manager.exception 121 self.assertEqual(13, proc_err.returncode) 122 self.assertEqual(error_cmd, proc_err.cmd) 123 124 def test_change_working_directory(self): 125 """Tests that cwd argument can be passed to subprocess.Popen().""" 126 tmp_dir = tempfile.mkdtemp(prefix='cmd_utils_test') 127 result = cmd_utils.run_command('pwd', shell=True, 128 read_stdout=True, raise_on_error=False, 129 cwd=tmp_dir) 130 self.assertEqual('%s\n' % tmp_dir, result.stdoutdata) 131 shutil.rmtree(tmp_dir) 132 133 if __name__ == '__main__': 134 logging.basicConfig(format='%(message)s', level=logging.INFO) 135 unittest.main() 136
