aboutsummaryrefslogtreecommitdiffstats
path: root/xtesting
diff options
context:
space:
mode:
Diffstat (limited to 'xtesting')
-rw-r--r--xtesting/core/feature.py28
-rw-r--r--xtesting/tests/unit/core/test_feature.py111
2 files changed, 117 insertions, 22 deletions
diff --git a/xtesting/core/feature.py b/xtesting/core/feature.py
index 2cb36bd6..2730179f 100644
--- a/xtesting/core/feature.py
+++ b/xtesting/core/feature.py
@@ -15,7 +15,9 @@ helpers to run any python method or any bash command.
import abc
import logging
+import os
import subprocess
+import sys
import time
import six
@@ -86,8 +88,8 @@ class BashFeature(Feature):
def __init__(self, **kwargs):
super(BashFeature, self).__init__(**kwargs)
- dir_results = "/var/lib/xtesting/results"
- self.result_file = "{}/{}.log".format(dir_results, self.case_name)
+ self.res_dir = "/var/lib/xtesting/results/{}".format(self.case_name)
+ self.result_file = "{}/{}.log".format(self.res_dir, self.case_name)
def execute(self, **kwargs):
"""Execute the cmd passed as arg
@@ -101,12 +103,22 @@ class BashFeature(Feature):
"""
try:
cmd = kwargs["cmd"]
- with open(self.result_file, 'w+') as f_stdout:
- subprocess.check_call(
- cmd, shell=True, stdout=f_stdout, stderr=subprocess.STDOUT)
- return 0
+ console = kwargs["console"] if "console" in kwargs else False
+ if not os.path.isdir(self.res_dir):
+ os.makedirs(self.res_dir)
+ with open(self.result_file, 'w') as f_stdout:
+ self.__logger.info("Calling %s", cmd)
+ process = subprocess.Popen(
+ cmd, shell=True, stdout=subprocess.PIPE,
+ stderr=subprocess.STDOUT)
+ for line in iter(process.stdout.readline, ''):
+ if console:
+ sys.stdout.write(line)
+ f_stdout.write(line)
+ process.wait()
+ with open(self.result_file, 'r') as f_stdin:
+ self.__logger.debug("$ %s\n%s", cmd, f_stdin.read().rstrip())
+ return process.returncode
except KeyError:
self.__logger.error("Please give cmd as arg. kwargs: %s", kwargs)
- except subprocess.CalledProcessError:
- self.__logger.error("Execute command: %s failed", cmd)
return -1
diff --git a/xtesting/tests/unit/core/test_feature.py b/xtesting/tests/unit/core/test_feature.py
index 47766cdd..cbee8511 100644
--- a/xtesting/tests/unit/core/test_feature.py
+++ b/xtesting/tests/unit/core/test_feature.py
@@ -14,6 +14,7 @@ import subprocess
import unittest
import mock
+import six
from xtesting.core import feature
from xtesting.core import testcase
@@ -38,7 +39,7 @@ class FeatureTestingBase(unittest.TestCase):
_project_name = "bar"
_repo = "dir_repo_bar"
_cmd = "run_bar_tests.py"
- _output_file = '/var/lib/xtesting/results/foo.log'
+ _output_file = '/var/lib/xtesting/results/foo/foo.log'
feature = None
@mock.patch('time.time', side_effect=[1, 2])
@@ -52,6 +53,15 @@ class FeatureTestingBase(unittest.TestCase):
self.assertEqual(self.feature.start_time, 1)
self.assertEqual(self.feature.stop_time, 2)
+ @mock.patch('time.time', side_effect=[1, 2])
+ def _test_run_console(self, console, status, mock_method=None):
+ self.assertEqual(
+ self.feature.run(cmd=self._cmd, console=console), status)
+ self.assertEqual(self.feature.result, 100)
+ mock_method.assert_has_calls([mock.call(), mock.call()])
+ self.assertEqual(self.feature.start_time, 1)
+ self.assertEqual(self.feature.stop_time, 2)
+
class FeatureTesting(FeatureTestingBase):
@@ -85,31 +95,104 @@ class BashFeatureTesting(FeatureTestingBase):
self.feature = feature.BashFeature(
project_name=self._project_name, case_name=self._case_name)
- @mock.patch('subprocess.check_call')
+ @mock.patch('subprocess.Popen')
def test_run_no_cmd(self, mock_subproc):
- delattr(FeatureTesting, "_cmd")
self.assertEqual(
self.feature.run(), testcase.TestCase.EX_RUN_ERROR)
mock_subproc.assert_not_called()
- @mock.patch('subprocess.check_call',
+ @mock.patch('os.path.isdir', return_value=True)
+ @mock.patch('subprocess.Popen',
side_effect=subprocess.CalledProcessError(0, '', ''))
- def test_run_ko(self, mock_subproc):
- setattr(FeatureTesting, "_cmd", "run_bar_tests.py")
+ def test_run_ko1(self, *args):
with mock.patch('six.moves.builtins.open', mock.mock_open()) as mopen:
self._test_run(testcase.TestCase.EX_RUN_ERROR)
- mopen.assert_called_once_with(self._output_file, "w+")
- mock_subproc.assert_called_once_with(
+ mopen.assert_called_once_with(self._output_file, "w")
+ args[0].assert_called_once_with(
self._cmd, shell=True, stderr=mock.ANY, stdout=mock.ANY)
-
- @mock.patch('subprocess.check_call')
- def test_run(self, mock_subproc):
- setattr(FeatureTesting, "_cmd", "run_bar_tests.py")
+ args[1].assert_called_once_with(self.feature.res_dir)
+
+ @mock.patch('os.path.isdir', return_value=True)
+ @mock.patch('subprocess.Popen')
+ def test_run_ko2(self, *args):
+ stream = six.StringIO()
+ stream.write("foo")
+ stream.seek(0)
+ attrs = {'return_value.stdout': stream, 'return_value.returncode': 1}
+ args[0].configure_mock(**attrs)
+ with mock.patch('six.moves.builtins.open', mock.mock_open()) as mopen:
+ self._test_run(testcase.TestCase.EX_RUN_ERROR)
+ self.assertIn(mock.call(self._output_file, 'w'), mopen.mock_calls)
+ self.assertIn(mock.call(self._output_file, 'r'), mopen.mock_calls)
+ args[0].assert_called_once_with(
+ self._cmd, shell=True, stderr=mock.ANY, stdout=mock.ANY)
+ args[1].assert_called_once_with(self.feature.res_dir)
+
+ @mock.patch('os.path.isdir', return_value=True)
+ @mock.patch('subprocess.Popen')
+ def test_run1(self, *args):
+ stream = six.StringIO()
+ stream.write("foo")
+ stream.seek(0)
+ attrs = {'return_value.stdout': stream, 'return_value.returncode': 0}
+ args[0].configure_mock(**attrs)
with mock.patch('six.moves.builtins.open', mock.mock_open()) as mopen:
self._test_run(testcase.TestCase.EX_OK)
- mopen.assert_called_once_with(self._output_file, "w+")
- mock_subproc.assert_called_once_with(
+ self.assertIn(mock.call(self._output_file, 'w'), mopen.mock_calls)
+ self.assertIn(mock.call(self._output_file, 'r'), mopen.mock_calls)
+ args[0].assert_called_once_with(
+ self._cmd, shell=True, stderr=mock.ANY, stdout=mock.ANY)
+ args[1].assert_called_once_with(self.feature.res_dir)
+
+ @mock.patch('os.path.isdir', return_value=True)
+ @mock.patch('subprocess.Popen')
+ def test_run2(self, *args):
+ stream = six.StringIO()
+ stream.write("foo")
+ stream.seek(0)
+ attrs = {'return_value.stdout': stream, 'return_value.returncode': 0}
+ args[0].configure_mock(**attrs)
+ with mock.patch('six.moves.builtins.open', mock.mock_open()) as mopen:
+ self._test_run_console(True, testcase.TestCase.EX_OK)
+ self.assertIn(mock.call(self._output_file, 'w'), mopen.mock_calls)
+ self.assertIn(mock.call(self._output_file, 'r'), mopen.mock_calls)
+ args[0].assert_called_once_with(
+ self._cmd, shell=True, stderr=mock.ANY, stdout=mock.ANY)
+ args[1].assert_called_once_with(self.feature.res_dir)
+
+ @mock.patch('os.path.isdir', return_value=True)
+ @mock.patch('subprocess.Popen')
+ def test_run3(self, *args):
+ stream = six.StringIO()
+ stream.write("foo")
+ stream.seek(0)
+ attrs = {'return_value.stdout': stream, 'return_value.returncode': 0}
+ args[0].configure_mock(**attrs)
+ with mock.patch('six.moves.builtins.open', mock.mock_open()) as mopen:
+ self._test_run_console(False, testcase.TestCase.EX_OK)
+ self.assertIn(mock.call(self._output_file, 'w'), mopen.mock_calls)
+ self.assertIn(mock.call(self._output_file, 'r'), mopen.mock_calls)
+ args[0].assert_called_once_with(
+ self._cmd, shell=True, stderr=mock.ANY, stdout=mock.ANY)
+ args[1].assert_called_once_with(self.feature.res_dir)
+
+ @mock.patch('os.makedirs')
+ @mock.patch('os.path.isdir', return_value=False)
+ @mock.patch('subprocess.Popen')
+ def test_run4(self, *args):
+ stream = six.StringIO()
+ stream.write("foo")
+ stream.seek(0)
+ attrs = {'return_value.stdout': stream, 'return_value.returncode': 0}
+ args[0].configure_mock(**attrs)
+ with mock.patch('six.moves.builtins.open', mock.mock_open()) as mopen:
+ self._test_run_console(False, testcase.TestCase.EX_OK)
+ self.assertIn(mock.call(self._output_file, 'w'), mopen.mock_calls)
+ self.assertIn(mock.call(self._output_file, 'r'), mopen.mock_calls)
+ args[0].assert_called_once_with(
self._cmd, shell=True, stderr=mock.ANY, stdout=mock.ANY)
+ args[1].assert_called_once_with(self.feature.res_dir)
+ args[2].assert_called_once_with(self.feature.res_dir)
if __name__ == "__main__":