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 """Implement a Xtesting driver to run mts suite."""
  13
  14 import csv
  15 import logging
  16 import os
  17 import shutil
  18 import time
  19
  20 from lxml import etree
  21 import prettytable
  22
  23 from xtesting.core import feature
  24 from xtesting.core import testcase
  25
  26
  27 __author__ = ("Vincent Mahe <v.mahe@orange.com>, "
  28               "Cedric Ollivier <cedric.ollivier@orange.com>")
  29
  30
  31 class MTSLauncher(feature.BashFeature):
  32     """Class designed to run MTS tests."""
  33
  34     __logger = logging.getLogger(__name__)
  35     mts_install_dir = "/opt/mts"
  36
  37     def check_requirements(self):
  38         """Check if startCmd.sh is in /opt/mts/bin"""
  39         if not os.path.exists(
  40                 os.path.join(self.mts_install_dir, 'bin/startCmd.sh')):
  41             self.__logger.warning(
  42                 "mts is not available for arm for the time being")
  43             self.is_skipped = True
  44
  45     def __init__(self, **kwargs):
  46         super().__init__(**kwargs)
  47         # Location of the HTML report generated by MTS
  48         self.mts_stats_dir = os.path.join(self.res_dir, 'mts_stats_report')
  49         # Location of the log files generated by MTS for each test.
  50         # Need to end path with a separator because of a bug in MTS.
  51         self.mts_logs_dir = os.path.join(self.res_dir,
  52                                          'mts_logs' + os.path.sep)
  53         # The location of file named testPlan.csv
  54         # that it always in $MTS_HOME/logs
  55         self.mts_result_csv_file = self.mts_install_dir + os.path.sep
  56         self.mts_result_csv_file += ("logs" + os.path.sep + "testPlan.csv")
  57         self.total_tests = 0
  58         self.pass_tests = 0
  59         self.fail_tests = 0
  60         self.skip_tests = 0
  61         self.response = None
  62         self.testcases = []
  63
  64     def parse_results(self):
  65         """Parse testPlan.csv containing the status of each testcase of the test file.
  66         See sample file in `xtesting/samples/mts/output/testPlan.csv`
  67         """
  68         with open(self.mts_result_csv_file) as stream_:
  69             self.__logger.info("Parsing file : %s", self.mts_result_csv_file)
  70             reader = csv.reader(stream_, delimiter=';')
  71             rownum = 0
  72             _tests_data = []
  73             msg = prettytable.PrettyTable(
  74                 header_style='upper', padding_width=5,
  75                 field_names=['MTS test', 'MTS test case',
  76                              'status'])
  77             for row in reader:
  78                 _test_dict = {}
  79                 nb_values = len(row)
  80                 if rownum > 0:
  81                     # If there's only one delimiter,
  82                     # it is the name of the <test> elt
  83                     if nb_values == 2:
  84                         test_name = row[0]
  85                         _test_dict['parent'] = test_name
  86                     elif nb_values == 3:
  87                         testcase_name = row[0].lstrip()
  88                         testcase_status = row[2]
  89                         self.total_tests += 1
  90                         if testcase_status == 'OK':
  91                             self.pass_tests += 1
  92                         elif testcase_status == 'Failed':
  93                             self.fail_tests += 1
  94                         elif testcase_status == '?':
  95                             self.skip_tests += 1
  96                         _test_dict['status'] = testcase_status
  97                         _test_dict['name'] = testcase_name
  98                         msg.add_row(
  99                             [test_name,
 100                              _test_dict['name'],
 101                              _test_dict['status']])
 102                 rownum += 1
 103                 _tests_data.append(_test_dict)
 104             try:
 105                 self.result = 100 * (
 106                     self.pass_tests / self.total_tests)
 107             except ZeroDivisionError:
 108                 self.__logger.error("No test has been run")
 109             self.__logger.info("MTS Test result:\n\n%s\n", msg.get_string())
 110             self.details = {}
 111             self.details['description'] = "Execution of some MTS tests"
 112             self.details['total_tests'] = self.total_tests
 113             self.details['pass_tests'] = self.pass_tests
 114             self.details['fail_tests'] = self.fail_tests
 115             self.details['skip_tests'] = self.skip_tests
 116             self.details['tests'] = _tests_data
 117
 118     def parse_xml_test_file(self, xml_test_file):
 119         """Parse the XML file containing the test definition for MTS.
 120         See sample file in `xtesting/samples/mts/test.xml`
 121         """
 122         nb_testcases = -1
 123         self.__logger.info(
 124             "Parsing XML test file %s containing the MTS tests definitions.",
 125             xml_test_file)
 126         try:
 127             parser = etree.XMLParser(load_dtd=True, resolve_entities=True)
 128             self.__logger.info("XML test file %s successfully parsed.",
 129                                xml_test_file)
 130             root = etree.parse(xml_test_file, parser=parser)
 131             # Need to look at all child nodes because there may be
 132             # some <for> elt between <test> and <testcase> elt
 133             self.testcases = root.xpath('//test//testcase/@name')
 134             nb_testcases = len(self.testcases)
 135             if nb_testcases == 0:
 136                 self.__logger.warning("Found no MTS testcase !")
 137             elif nb_testcases == 1:
 138                 self.__logger.info("Found only one MTS testcase: %s",
 139                                    self.testcases[0])
 140             else:
 141                 self.__logger.info("Found %d MTS testcases :", nb_testcases)
 142                 for mts_testcase in self.testcases:
 143                     self.__logger.info("    - %s", mts_testcase)
 144         except etree.XMLSyntaxError as xml_err:
 145             self.__logger.error("Error while parsing XML test file: %s",
 146                                 str(xml_err))
 147         return nb_testcases
 148
 149     def check_enabled_mts_test_cases(self, enabled_testcases):
 150         """Make sure that all required MTS test cases exist
 151         in the XML test file.
 152         """
 153         if enabled_testcases:
 154             # Verify if the MTS test case exists in the whole list of test
 155             # cases declared in the test XML file
 156             for enabled_testcase in enabled_testcases:
 157                 if enabled_testcase not in self.testcases:
 158                     self.__logger.error(
 159                         "The required MTS testcase named `%s` does not exist"
 160                         " !", enabled_testcase)
 161                     return False
 162         return True
 163
 164     def execute(self, **kwargs):  # pylint: disable=too-many-locals
 165         try:
 166             # Read specific parameters for MTS
 167             test_file = kwargs["test_file"]
 168             log_level = kwargs[
 169                 "log_level"] if "log_level" in kwargs else "INFO"
 170             store_method = kwargs[
 171                 "store_method"] if "store_method" in kwargs else "FILE"
 172             # Must use the $HOME_MTS/bin as current working dir
 173             cwd = self.mts_install_dir + os.path.sep + "bin"
 174
 175             # Get the list of enabled MTS testcases, if any
 176             enabled_testcases = kwargs[
 177                 "testcases"] if "testcases" in kwargs else []
 178             enabled_testcases_str = ''
 179             if enabled_testcases:
 180                 enabled_testcases_str = ' '.join(enabled_testcases)
 181                 check_ok = self.check_enabled_mts_test_cases(enabled_testcases)
 182                 if not check_ok:
 183                     return -3
 184
 185             # Build command line to launch for MTS
 186             cmd = ("cd {} && ./startCmd.sh {} {} -sequential -levelLog:{}"
 187                    " -storageLog:{}"
 188                    " -config:stats.REPORT_DIRECTORY+{}"
 189                    " -config:logs.STORAGE_DIRECTORY+{}"
 190                    " -genReport:true"
 191                    " -showRep:false").format(cwd,
 192                                              test_file,
 193                                              enabled_testcases_str,
 194                                              log_level,
 195                                              store_method,
 196                                              self.mts_stats_dir,
 197                                              self.mts_logs_dir)
 198
 199             # Make sure to create the necessary output sub-folders for MTS
 200             # and cleanup output files from previous run.
 201             if os.path.exists(self.mts_result_csv_file):
 202                 os.remove(self.mts_result_csv_file)
 203
 204             if os.path.isdir(self.mts_stats_dir):
 205                 shutil.rmtree(self.mts_stats_dir)
 206             os.makedirs(self.mts_stats_dir)
 207
 208             if os.path.isdir(self.mts_logs_dir):
 209                 shutil.rmtree(self.mts_logs_dir)
 210             os.makedirs(self.mts_logs_dir)
 211
 212             self.__logger.info(
 213                 "MTS statistics output dir: %s ", self.mts_stats_dir)
 214             self.__logger.info("MTS logs output dir: %s ", self.mts_logs_dir)
 215
 216             kwargs.pop("cmd", None)
 217             return super().execute(cmd=cmd, **kwargs)
 218
 219         except KeyError:
 220             self.__logger.error("Missing mandatory arg for MTS. kwargs: %s",
 221                                 kwargs)
 222         return -1
 223
 224     def run(self, **kwargs):
 225         """Runs the MTS suite"""
 226         self.start_time = time.time()
 227         exit_code = testcase.TestCase.EX_RUN_ERROR
 228         self.result = 0
 229         try:
 230             nb_testcases = self.parse_xml_test_file(kwargs["test_file"])
 231             # Do something only if there are some MTS test cases in the test
 232             # file
 233             if nb_testcases > 0:
 234                 if self.execute(**kwargs) == 0:
 235                     exit_code = testcase.TestCase.EX_OK
 236                     try:
 237                         self.parse_results()
 238                     except Exception:  # pylint: disable=broad-except
 239                         self.__logger.exception(
 240                             "Cannot parse result file "
 241                             "$MTS_HOME/logs/testPlan.csv")
 242                         exit_code = testcase.TestCase.EX_RUN_ERROR
 243         except Exception:  # pylint: disable=broad-except
 244             self.__logger.exception("%s FAILED", self.project_name)
 245         self.stop_time = time.time()
 246         return exit_code