Returns:
0 if cmd returns 0,
- -1 otherwise.
+ non-zero in all other cases.
"""
try:
cmd = kwargs["cmd"]
console = kwargs["console"] if "console" in kwargs else False
+ # For some tests, we may need to force stop after N sec
+ max_duration = kwargs.get("max_duration")
if not os.path.isdir(self.res_dir):
os.makedirs(self.res_dir)
with open(self.result_file, 'w') as f_stdout:
if console:
sys.stdout.write(line.decode("utf-8"))
f_stdout.write(line.decode("utf-8"))
- process.wait()
+ try:
+ process.wait(timeout=max_duration)
+ except subprocess.TimeoutExpired:
+ process.kill()
+ self.__logger.info(
+ "Killing process after %d second(s).", max_duration)
+ return -2
with open(self.result_file, 'r') as f_stdin:
self.__logger.debug("$ %s\n%s", cmd, f_stdin.read().rstrip())
return process.returncode
import logging
import os
import shutil
-import subprocess
-import sys
import time
from lxml import etree
import prettytable
+from xtesting.core import feature
from xtesting.core import testcase
"Cedric Ollivier <cedric.ollivier@orange.com>")
-class MTSLauncher(testcase.TestCase):
+class MTSLauncher(feature.BashFeature):
"""Class designed to run MTS tests."""
__logger = logging.getLogger(__name__)
def __init__(self, **kwargs):
super(MTSLauncher, self).__init__(**kwargs)
- self.result_file = "{}/{}.log".format(self.res_dir, self.case_name)
# Location of the HTML report generated by MTS
self.mts_stats_dir = os.path.join(self.res_dir, 'mts_stats_report')
# Location of the log files generated by MTS for each test.
return True
def execute(self, **kwargs): # pylint: disable=too-many-locals
- """Execute the cmd passed as arg
-
- Args:
- kwargs: Arbitrary keyword arguments.
-
- Returns:
- 0 if cmd returns 0,
- -1 otherwise.
- """
try:
- console = kwargs["console"] if "console" in kwargs else False
# Read specific parameters for MTS
test_file = kwargs["test_file"]
log_level = kwargs[
"log_level"] if "log_level" in kwargs else "INFO"
-
- # For some MTS tests, we need to force stop after N sec
- max_duration = kwargs[
- "max_duration"] if "max_duration" in kwargs else None
store_method = kwargs[
"store_method"] if "store_method" in kwargs else "FILE"
# Must use the $HOME_MTS/bin as current working dir
enabled_testcases_str = ' '.join(enabled_testcases)
check_ok = self.check_enabled_mts_test_cases(enabled_testcases)
if not check_ok:
- return -2
+ return -3
# Build command line to launch for MTS
cmd = ("cd {} && ./startCmd.sh {} {} -sequential -levelLog:{}"
"MTS statistics output dir: %s ", self.mts_stats_dir)
self.__logger.info("MTS logs output dir: %s ", self.mts_logs_dir)
- # Launch MTS as a sub-process
- # and save its standard output to a file
- 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, b''):
- if console:
- sys.stdout.write(line.decode("utf-8"))
- f_stdout.write(line.decode("utf-8"))
- try:
- process.wait(timeout=max_duration)
- except subprocess.TimeoutExpired:
- process.kill()
- self.__logger.info(
- "Killing MTS process after %d second(s).",
- max_duration)
- return 3
- with open(self.result_file, 'r') as f_stdin:
- self.__logger.debug("$ %s\n%s", cmd, f_stdin.read().rstrip())
- return process.returncode
+ kwargs.pop("cmd", None)
+ return super(MTSLauncher, self).execute(cmd=cmd, **kwargs)
+
except KeyError:
self.__logger.error("Missing mandatory arg for MTS. kwargs: %s",
kwargs)
return -1
def run(self, **kwargs):
- """Run the feature.
-
- It allows executing any Python method by calling execute().
-
- It sets the following attributes required to push the results
- to DB:
-
- * result,
- * start_time,
- * stop_time.
-
- It doesn't fulfill details when pushing the results to the DB.
-
- Args:
- kwargs: Arbitrary keyword arguments.
-
- Returns:
- TestCase.EX_OK if execute() returns 0,
- TestCase.EX_RUN_ERROR otherwise.
- """
+ """Runs the MTS suite"""
self.start_time = time.time()
exit_code = testcase.TestCase.EX_RUN_ERROR
self.result = 0
_repo = "dir_repo_bar"
_cmd = "run_bar_tests.py"
_output_file = os.path.join(constants.RESULTS_DIR, 'foo/foo.log')
+ _max_duration = 1
feature = None
@mock.patch('time.time', side_effect=[1, 2])
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_max_duration(self, status, mock_method=None):
+ self.assertEqual(
+ self.feature.run(cmd=self._cmd, max_duration=self._max_duration),
+ status)
+ 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):
self._cmd, shell=True, stderr=mock.ANY, stdout=mock.ANY)
args[1].assert_called_once_with(self.feature.res_dir)
+ @mock.patch('subprocess.Popen')
+ @mock.patch('os.path.isdir', return_value=True)
+ def test_run_ko3(self, *args):
+ stream = BytesIO()
+ stream.write(b"foo")
+ stream.seek(0)
+ wait = mock.MagicMock(side_effect=subprocess.TimeoutExpired(
+ cmd=FeatureTestingBase._cmd,
+ timeout=FeatureTestingBase._max_duration))
+ kill = mock.MagicMock()
+ attrs = {'return_value.wait': wait,
+ 'return_value.kill': kill,
+ 'return_value.stdout': stream,
+ 'return_value.returncode': 0}
+ args[1].configure_mock(**attrs)
+ with mock.patch('builtins.open', mock.mock_open()) as mopen:
+ self._test_run_max_duration(testcase.TestCase.EX_RUN_ERROR)
+ self.assertIn(mock.call(self._output_file, 'w'), mopen.mock_calls)
+ self.assertNotIn(mock.call(self._output_file, 'r'), mopen.mock_calls)
+ args[1].assert_called_once_with(
+ self._cmd, shell=True, stderr=mock.ANY, stdout=mock.ANY)
+ wait.assert_called_once_with(timeout=FeatureTestingBase._max_duration)
+ kill.assert_called_once()
+ args[0].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):
Code Review / functest-xtesting.git / commitdiff