xtesting/core/mts.py

   1 #!/usr/bin/env python
   2
   3 # Copyright (c) 2020 Orange and others.
   4 #
   5 # All rights reserved. This program and the accompanying materials
   6 # are made available under the terms of the Apache License, Version 2.0
   7 # which accompanies this distribution, and is available at
   8 # http://www.apache.org/licenses/LICENSE-2.0
   9
  10 # pylint: disable=too-many-instance-attributes
  11
  12 """Define the parent classes of all Xtesting Features.
  13
  14 Feature is considered as TestCase offered by Third-party. It offers
  15 helpers to run any python method or any bash command.
  16 """
  17
  18 import csv
  19 import logging
  20 import os
  21 import shutil
  22 import subprocess
  23 import sys
  24 import time
  25
  26 from lxml import etree
  27 import prettytable
  28 import six
  29
  30 from xtesting.core import testcase
  31
  32
  33 __author__ = ("Vincent Mahe <v.mahe@orange.com>, "
  34               "Cedric Ollivier <cedric.ollivier@orange.com>")
  35
  36
  37 class MTSLauncher(testcase.TestCase):
  38     """Class designed to run MTS tests."""
  39
  40     __logger = logging.getLogger(__name__)
  41     mts_install_dir = "/opt/mts"
  42
  43     def check_requirements(self):
  44         """Check if startCmd.sh is in /opt/mts/bin"""
  45         if not os.path.exists(
  46                 os.path.join(self.mts_install_dir, 'bin/startCmd.sh')):
  47             self.__logger.warning(
  48                 "mts is not available for arm for the time being")
  49             self.is_skipped = True
  50
  51     def __init__(self, **kwargs):
  52         super(MTSLauncher, self).__init__(**kwargs)
  53         self.result_file = "{}/{}.log".format(self.res_dir, self.case_name)
  54         # Location of the HTML report generated by MTS
  55         self.mts_stats_dir = os.path.join(self.res_dir, 'mts_stats_report')
  56         # Location of the log files generated by MTS for each test.
  57         # Need to end path with a separator because of a bug in MTS.
  58         self.mts_logs_dir = os.path.join(self.res_dir,
  59                                          'mts_logs' + os.path.sep)
  60         # The location of file named testPlan.csv
  61         # that it always in $MTS_HOME/logs
  62         self.mts_result_csv_file = self.mts_install_dir + os.path.sep
  63         self.mts_result_csv_file += ("logs" + os.path.sep + "testPlan.csv")
  64         self.total_tests = 0
  65         self.pass_tests = 0
  66         self.fail_tests = 0
  67         self.skip_tests = 0
  68         self.response = None
  69         self.testcases = []
  70
  71     def parse_results(self):
  72         """Parse testPlan.csv containing the status of each testcase of the test file.
  73         See sample file in `xtesting/samples/mts/output/testPlan.csv`
  74         """
  75         with open(self.mts_result_csv_file) as stream_:
  76             self.__logger.info("Parsing file : %s", self.mts_result_csv_file)
  77             reader = csv.reader(stream_, delimiter=';')
  78             rownum = 0
  79             _tests_data = []
  80             msg = prettytable.PrettyTable(
  81                 header_style='upper', padding_width=5,
  82                 field_names=['MTS test', 'MTS test case',
  83                              'status'])
  84             for row in reader:
  85                 _test_dict = {}
  86                 nb_values = len(row)
  87                 if rownum > 0:
  88                     # If there's only one delimiter,
  89                     # it is the name of the <test> elt
  90                     if nb_values == 2:
  91                         test_name = row[0]
  92                         _test_dict['parent'] = test_name
  93                     elif nb_values == 3:
  94                         testcase_name = row[0].lstrip()
  95                         testcase_status = row[2]
  96                         self.total_tests += 1
  97                         if testcase_status == 'OK':
  98                             self.pass_tests += 1
  99                         elif testcase_status == 'Failed':
 100                             self.fail_tests += 1
 101                         elif testcase_status == '?':
 102                             self.skip_tests += 1
 103                         _test_dict['status'] = testcase_status
 104                         _test_dict['name'] = testcase_name
 105                         msg.add_row(
 106                             [test_name,
 107                              _test_dict['name'],
 108                              _test_dict['status']])
 109                 rownum += 1
 110                 _tests_data.append(_test_dict)
 111             try:
 112                 self.result = 100 * (
 113                     self.pass_tests / self.total_tests)
 114             except ZeroDivisionError:
 115                 self.__logger.error("No test has been run")
 116             self.__logger.info("MTS Test result:\n\n%s\n", msg.get_string())
 117             self.details = {}
 118             self.details['description'] = "Execution of some MTS tests"
 119             self.details['total_tests'] = self.total_tests
 120             self.details['pass_tests'] = self.pass_tests
 121             self.details['fail_tests'] = self.fail_tests
 122             self.details['skip_tests'] = self.skip_tests
 123             self.details['tests'] = _tests_data
 124
 125     def parse_xml_test_file(self, xml_test_file):
 126         """Parse the XML file containing the test definition for MTS.
 127         See sample file in `xtesting/samples/mts/test.xml`
 128         """
 129         nb_testcases = -1
 130         self.__logger.info(
 131             "Parsing XML test file %s containing the MTS tests definitions.",
 132             xml_test_file)
 133         try:
 134             parser = etree.XMLParser(load_dtd=True, resolve_entities=True)
 135             self.__logger.info("XML test file %s successfully parsed.",
 136                                xml_test_file)
 137             root = etree.parse(xml_test_file, parser=parser)
 138             # Need to look at all child nodes because there may be
 139             # some <for> elt between <test> and <testcase> elt
 140             self.testcases = root.xpath('//test//testcase/@name')
 141             nb_testcases = len(self.testcases)
 142             if nb_testcases == 0:
 143                 self.__logger.warning("Found no MTS testcase !")
 144             elif nb_testcases == 1:
 145                 self.__logger.info("Found only one MTS testcase: %s",
 146                                    self.testcases[0])
 147             else:
 148                 self.__logger.info("Found %d MTS testcases :", nb_testcases)
 149                 for mts_testcase in self.testcases:
 150                     self.__logger.info("    - %s", mts_testcase)
 151         except etree.XMLSyntaxError as xml_err:
 152             self.__logger.error("Error while parsing XML test file: %s",
 153                                 str(xml_err))
 154         return nb_testcases
 155
 156     def check_enabled_mts_test_cases(self, enabled_testcases):
 157         """Make sure that all required MTS test cases exist
 158         in the XML test file.
 159         """
 160         if enabled_testcases:
 161             # Verify if the MTS test case exists in the whole list of test
 162             # cases declared in the test XML file
 163             for enabled_testcase in enabled_testcases:
 164                 if enabled_testcase not in self.testcases:
 165                     self.__logger.error(
 166                         "The required MTS testcase named `%s` does not exist"
 167                         " !", enabled_testcase)
 168                     return False
 169         return True
 170
 171     def execute(self, **kwargs):  # pylint: disable=too-many-locals
 172         """Execute the cmd passed as arg
 173
 174         Args:
 175             kwargs: Arbitrary keyword arguments.
 176
 177         Returns:
 178             0 if cmd returns 0,
 179             -1 otherwise.
 180         """
 181         try:
 182             console = kwargs["console"] if "console" in kwargs else False
 183             # Read specific parameters for MTS
 184             test_file = kwargs["test_file"]
 185             log_level = kwargs[
 186                 "log_level"] if "log_level" in kwargs else "INFO"
 187
 188             # For some MTS tests, we need to force stop after N sec
 189             max_duration = kwargs[
 190                 "max_duration"] if "max_duration" in kwargs else None
 191             store_method = kwargs[
 192                 "store_method"] if "store_method" in kwargs else "FILE"
 193             # Must use the $HOME_MTS/bin as current working dir
 194             cwd = self.mts_install_dir + os.path.sep + "bin"
 195
 196             # Get the list of enabled MTS testcases, if any
 197             enabled_testcases = kwargs[
 198                 "testcases"] if "testcases" in kwargs else []
 199             enabled_testcases_str = ''
 200             if enabled_testcases:
 201                 enabled_testcases_str = ' '.join(enabled_testcases)
 202                 check_ok = self.check_enabled_mts_test_cases(enabled_testcases)
 203                 if not check_ok:
 204                     return -2
 205
 206             # Build command line to launch for MTS
 207             cmd = ("cd {} && ./startCmd.sh {} {} -sequential -levelLog:{}"
 208                    " -storageLog:{}"
 209                    " -config:stats.REPORT_DIRECTORY+{}"
 210                    " -config:logs.STORAGE_DIRECTORY+{}"
 211                    " -genReport:true"
 212                    " -showRep:false").format(cwd,
 213                                              test_file,
 214                                              enabled_testcases_str,
 215                                              log_level,
 216                                              store_method,
 217                                              self.mts_stats_dir,
 218                                              self.mts_logs_dir)
 219
 220             # Make sure to create the necessary output sub-folders for MTS
 221             # and cleanup output files from previous run.
 222             if os.path.exists(self.mts_result_csv_file):
 223                 os.remove(self.mts_result_csv_file)
 224
 225             if os.path.isdir(self.mts_stats_dir):
 226                 shutil.rmtree(self.mts_stats_dir)
 227             os.makedirs(self.mts_stats_dir)
 228
 229             if os.path.isdir(self.mts_logs_dir):
 230                 shutil.rmtree(self.mts_logs_dir)
 231             os.makedirs(self.mts_logs_dir)
 232
 233             self.__logger.info(
 234                 "MTS statistics output dir: %s ", self.mts_stats_dir)
 235             self.__logger.info("MTS logs output dir: %s ", self.mts_logs_dir)
 236
 237             # Launch MTS as a sub-process
 238             # and save its standard output to a file
 239             with open(self.result_file, 'w') as f_stdout:
 240                 self.__logger.info("Calling %s", cmd)
 241                 process = subprocess.Popen(
 242                     cmd, shell=True, stdout=subprocess.PIPE,
 243                     stderr=subprocess.STDOUT)
 244                 for line in iter(process.stdout.readline, b''):
 245                     if console:
 246                         sys.stdout.write(line.decode("utf-8"))
 247                     f_stdout.write(line.decode("utf-8"))
 248                 if six.PY3:
 249                     try:
 250                         process.wait(timeout=max_duration)
 251                     except subprocess.TimeoutExpired:
 252                         process.kill()
 253                         self.__logger.info(
 254                             "Killing MTS process after %d second(s).",
 255                             max_duration)
 256                         return 3
 257                 else:
 258                     process.wait()
 259             with open(self.result_file, 'r') as f_stdin:
 260                 self.__logger.debug("$ %s\n%s", cmd, f_stdin.read().rstrip())
 261             return process.returncode
 262         except KeyError:
 263             self.__logger.error("Missing mandatory arg for MTS. kwargs: %s",
 264                                 kwargs)
 265         return -1
 266
 267     def run(self, **kwargs):
 268         """Run the feature.
 269
 270         It allows executing any Python method by calling execute().
 271
 272         It sets the following attributes required to push the results
 273         to DB:
 274
 275             * result,
 276             * start_time,
 277             * stop_time.
 278
 279         It doesn't fulfill details when pushing the results to the DB.
 280
 281         Args:
 282             kwargs: Arbitrary keyword arguments.
 283
 284         Returns:
 285             TestCase.EX_OK if execute() returns 0,
 286             TestCase.EX_RUN_ERROR otherwise.
 287         """
 288         self.start_time = time.time()
 289         exit_code = testcase.TestCase.EX_RUN_ERROR
 290         self.result = 0
 291         try:
 292             nb_testcases = self.parse_xml_test_file(kwargs["test_file"])
 293             # Do something only if there are some MTS test cases in the test
 294             # file
 295             if nb_testcases > 0:
 296                 if self.execute(**kwargs) == 0:
 297                     exit_code = testcase.TestCase.EX_OK
 298                     try:
 299                         self.parse_results()
 300                     except Exception:  # pylint: disable=broad-except
 301                         self.__logger.exception(
 302                             "Cannot parse result file "
 303                             "$MTS_HOME/logs/testPlan.csv")
 304                         exit_code = testcase.TestCase.EX_RUN_ERROR
 305         except Exception:  # pylint: disable=broad-except
 306             self.__logger.exception("%s FAILED", self.project_name)
 307         self.stop_time = time.time()
 308         return exit_code