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