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