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, encoding='utf-8') 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 = (f"cd {cwd} && ./startCmd.sh {test_file} "
 187                    f"{enabled_testcases_str} -sequential -levelLog:{log_level}"
 188                    f" -storageLog:{store_method}"
 189                    f" -config:stats.REPORT_DIRECTORY+{self.mts_stats_dir}"
 190                    f" -config:logs.STORAGE_DIRECTORY+{self.mts_logs_dir}"
 191                    " -genReport:true"
 192                    " -showRep:false")
 193
 194             # Make sure to create the necessary output sub-folders for MTS
 195             # and cleanup output files from previous run.
 196             if os.path.exists(self.mts_result_csv_file):
 197                 os.remove(self.mts_result_csv_file)
 198
 199             if os.path.isdir(self.mts_stats_dir):
 200                 shutil.rmtree(self.mts_stats_dir)
 201             os.makedirs(self.mts_stats_dir)
 202
 203             if os.path.isdir(self.mts_logs_dir):
 204                 shutil.rmtree(self.mts_logs_dir)
 205             os.makedirs(self.mts_logs_dir)
 206
 207             self.__logger.info(
 208                 "MTS statistics output dir: %s ", self.mts_stats_dir)
 209             self.__logger.info("MTS logs output dir: %s ", self.mts_logs_dir)
 210
 211             kwargs.pop("cmd", None)
 212             return super().execute(cmd=cmd, **kwargs)
 213
 214         except KeyError:
 215             self.__logger.error("Missing mandatory arg for MTS. kwargs: %s",
 216                                 kwargs)
 217         return -1
 218
 219     def run(self, **kwargs):
 220         """Runs the MTS suite"""
 221         self.start_time = time.time()
 222         exit_code = testcase.TestCase.EX_RUN_ERROR
 223         self.result = 0
 224         try:
 225             nb_testcases = self.parse_xml_test_file(kwargs["test_file"])
 226             # Do something only if there are some MTS test cases in the test
 227             # file
 228             if nb_testcases > 0:
 229                 if self.execute(**kwargs) == 0:
 230                     exit_code = testcase.TestCase.EX_OK
 231                     try:
 232                         self.parse_results()
 233                     except Exception:  # pylint: disable=broad-except
 234                         self.__logger.exception(
 235                             "Cannot parse result file "
 236                             "$MTS_HOME/logs/testPlan.csv")
 237                         exit_code = testcase.TestCase.EX_RUN_ERROR
 238         except Exception:  # pylint: disable=broad-except
 239             self.__logger.exception("%s FAILED", self.project_name)
 240         self.stop_time = time.time()
 241         return exit_code