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