Merge "Added missing test coverage for run_command_and_log function"

tripleoclient/tests/test_utils.py

@@ -16,9 +16,11 @@
 
 import argparse
 import datetime
+import logging
 import mock
 import os
 import os.path
+import subprocess
 import tempfile
 
 from heatclient import exc as hc_exc
@@ -161,6 +163,92 @@ class TestRunAnsiblePlaybook(TestCase):
                                          env=env, retcode_only=False)
 
 
+class TestRunCommandAndLog(TestCase):
+    def setUp(self):
+        self.mock_logger = mock.Mock(spec=logging.Logger)
+
+        self.mock_process = mock.Mock()
+        self.mock_process.stdout.readline.side_effect = ['foo\n', 'bar\n']
+        self.mock_process.wait.side_effect = [0]
+        self.mock_process.returncode = 0
+
+        mock_sub = mock.patch('subprocess.Popen',
+                              return_value=self.mock_process)
+        self.mock_popen = mock_sub.start()
+        self.addCleanup(mock_sub.stop)
+
+        self.cmd = ['exit', '0']
+        self.e_cmd = ['exit', '1']
+        self.log_calls = [mock.call('foo'),
+                          mock.call('bar')]
+
+    def test_success_default(self):
+        retcode = utils.run_command_and_log(self.mock_logger, self.cmd)
+        self.mock_popen.assert_called_once_with(self.cmd,
+                                                stdout=subprocess.PIPE,
+                                                stderr=subprocess.STDOUT,
+                                                shell=False, bufsize=1,
+                                                cwd=None, env=None)
+        self.assertEqual(retcode, 0)
+        self.mock_logger.warning.assert_has_calls(self.log_calls,
+                                                  any_order=False)
+
+    @mock.patch('subprocess.Popen')
+    def test_error_subprocess(self, mock_popen):
+        mock_process = mock.Mock()
+        mock_process.stdout.readline.side_effect = ['Error\n']
+        mock_process.wait.side_effect = [1]
+        mock_process.returncode = 1
+
+        mock_popen.return_value = mock_process
+
+        retcode = utils.run_command_and_log(self.mock_logger, self.e_cmd)
+        mock_popen.assert_called_once_with(self.e_cmd, stdout=subprocess.PIPE,
+                                           stderr=subprocess.STDOUT,
+                                           shell=False, bufsize=1, cwd=None,
+                                           env=None)
+
+        self.assertEqual(retcode, 1)
+        self.mock_logger.warning.assert_called_once_with('Error')
+
+    def test_success_env(self):
+        test_env = os.environ.copy()
+        retcode = utils.run_command_and_log(self.mock_logger, self.cmd,
+                                            env=test_env)
+        self.mock_popen.assert_called_once_with(self.cmd,
+                                                stdout=subprocess.PIPE,
+                                                stderr=subprocess.STDOUT,
+                                                shell=False, bufsize=1,
+                                                cwd=None, env=test_env)
+        self.assertEqual(retcode, 0)
+        self.mock_logger.warning.assert_has_calls(self.log_calls,
+                                                  any_order=False)
+
+    def test_success_cwd(self):
+        test_cwd = '/usr/local/bin'
+        retcode = utils.run_command_and_log(self.mock_logger, self.cmd,
+                                            cwd=test_cwd)
+        self.mock_popen.assert_called_once_with(self.cmd,
+                                                stdout=subprocess.PIPE,
+                                                stderr=subprocess.STDOUT,
+                                                shell=False, bufsize=1,
+                                                cwd=test_cwd, env=None)
+        self.assertEqual(retcode, 0)
+        self.mock_logger.warning.assert_has_calls(self.log_calls,
+                                                  any_order=False)
+
+    def test_success_no_retcode(self):
+        run = utils.run_command_and_log(self.mock_logger, self.cmd,
+                                        retcode_only=False)
+        self.mock_popen.assert_called_once_with(self.cmd,
+                                                stdout=subprocess.PIPE,
+                                                stderr=subprocess.STDOUT,
+                                                shell=False, bufsize=1,
+                                                cwd=None, env=None)
+        self.assertEqual(run, self.mock_process)
+        self.mock_logger.warning.assert_not_called()
+
+
 class TestWaitForStackUtil(TestCase):
     def setUp(self):
         self.mock_orchestration = mock.Mock()