3 # Copyright (c) 2020 Orange and others.
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
10 # pylint: disable=too-many-instance-attributes
12 """Define the parent classes of all Xtesting Features.
14 Feature is considered as TestCase offered by Third-party. It offers
15 helpers to run any python method or any bash command.
26 from lxml import etree
30 from xtesting.core import testcase
33 __author__ = ("Vincent Mahe <v.mahe@orange.com>, "
34 "Cedric Ollivier <cedric.ollivier@orange.com>")
37 class MTSLauncher(testcase.TestCase):
38 """Class designed to run MTS tests."""
40 __logger = logging.getLogger(__name__)
41 mts_install_dir = "/opt/mts"
43 def check_requirements(self):
44 """Check if startCmd.sh is in /opt/mts/bin"""
45 if not os.path.exists(
46 os.path.join(self.mts_install_dir, 'bin/startCmd.sh')):
47 self.__logger.warning(
48 "mts is not available for arm for the time being")
49 self.is_skipped = True
51 def __init__(self, **kwargs):
52 super(MTSLauncher, self).__init__(**kwargs)
53 self.result_file = "{}/{}.log".format(self.res_dir, self.case_name)
54 # Location of the HTML report generated by MTS
55 self.mts_stats_dir = os.path.join(self.res_dir, 'mts_stats_report')
56 # Location of the log files generated by MTS for each test.
57 # Need to end path with a separator because of a bug in MTS.
58 self.mts_logs_dir = os.path.join(self.res_dir,
59 'mts_logs' + os.path.sep)
60 # The location of file named testPlan.csv
61 # that it always in $MTS_HOME/logs
62 self.mts_result_csv_file = self.mts_install_dir + os.path.sep
63 self.mts_result_csv_file += ("logs" + os.path.sep + "testPlan.csv")
71 def parse_results(self):
72 """Parse testPlan.csv containing the status of each testcase of the test file.
73 See sample file in `xtesting/samples/mts/output/testPlan.csv`
75 with open(self.mts_result_csv_file) as stream_:
76 self.__logger.info("Parsing file : %s", self.mts_result_csv_file)
77 reader = csv.reader(stream_, delimiter=';')
80 msg = prettytable.PrettyTable(
81 header_style='upper', padding_width=5,
82 field_names=['MTS test', 'MTS test case',
88 # If there's only one delimiter,
89 # it is the name of the <test> elt
92 _test_dict['parent'] = test_name
94 testcase_name = row[0].lstrip()
95 testcase_status = row[2]
97 if testcase_status == 'OK':
99 elif testcase_status == 'Failed':
101 elif testcase_status == '?':
103 _test_dict['status'] = testcase_status
104 _test_dict['name'] = testcase_name
108 _test_dict['status']])
110 _tests_data.append(_test_dict)
112 self.result = 100 * (
113 self.pass_tests / self.total_tests)
114 except ZeroDivisionError:
115 self.__logger.error("No test has been run")
116 self.__logger.info("MTS Test result:\n\n%s\n", msg.get_string())
118 self.details['description'] = "Execution of some MTS tests"
119 self.details['total_tests'] = self.total_tests
120 self.details['pass_tests'] = self.pass_tests
121 self.details['fail_tests'] = self.fail_tests
122 self.details['skip_tests'] = self.skip_tests
123 self.details['tests'] = _tests_data
125 def parse_xml_test_file(self, xml_test_file):
126 """Parse the XML file containing the test definition for MTS.
127 See sample file in `xtesting/samples/mts/test.xml`
131 "Parsing XML test file %s containing the MTS tests definitions.",
134 parser = etree.XMLParser(load_dtd=True, resolve_entities=True)
135 self.__logger.info("XML test file %s successfully parsed.",
137 root = etree.parse(xml_test_file, parser=parser)
138 # Need to look at all child nodes because there may be
139 # some <for> elt between <test> and <testcase> elt
140 self.testcases = root.xpath('//test//testcase/@name')
141 nb_testcases = len(self.testcases)
142 if nb_testcases == 0:
143 self.__logger.warning("Found no MTS testcase !")
144 elif nb_testcases == 1:
145 self.__logger.info("Found only one MTS testcase: %s",
148 self.__logger.info("Found %d MTS testcases :", nb_testcases)
149 for mts_testcase in self.testcases:
150 self.__logger.info(" - %s", mts_testcase)
151 except etree.XMLSyntaxError as xml_err:
152 self.__logger.error("Error while parsing XML test file: %s",
156 def check_enabled_mts_test_cases(self, enabled_testcases):
157 """Make sure that all required MTS test cases exist
158 in the XML test file.
160 if enabled_testcases:
161 # Verify if the MTS test case exists in the whole list of test
162 # cases declared in the test XML file
163 for enabled_testcase in enabled_testcases:
164 if enabled_testcase not in self.testcases:
166 "The required MTS testcase named `%s` does not exist"
167 " !", enabled_testcase)
171 def execute(self, **kwargs): # pylint: disable=too-many-locals
172 """Execute the cmd passed as arg
175 kwargs: Arbitrary keyword arguments.
182 console = kwargs["console"] if "console" in kwargs else False
183 # Read specific parameters for MTS
184 test_file = kwargs["test_file"]
186 "log_level"] if "log_level" in kwargs else "INFO"
188 # For some MTS tests, we need to force stop after N sec
189 max_duration = kwargs[
190 "max_duration"] if "max_duration" in kwargs else None
191 store_method = kwargs[
192 "store_method"] if "store_method" in kwargs else "FILE"
193 # Must use the $HOME_MTS/bin as current working dir
194 cwd = self.mts_install_dir + os.path.sep + "bin"
196 # Get the list of enabled MTS testcases, if any
197 enabled_testcases = kwargs[
198 "testcases"] if "testcases" in kwargs else []
199 enabled_testcases_str = ''
200 if enabled_testcases:
201 enabled_testcases_str = ' '.join(enabled_testcases)
202 check_ok = self.check_enabled_mts_test_cases(enabled_testcases)
206 # Build command line to launch for MTS
207 cmd = ("cd {} && ./startCmd.sh {} {} -sequential -levelLog:{}"
209 " -config:stats.REPORT_DIRECTORY+{}"
210 " -config:logs.STORAGE_DIRECTORY+{}"
212 " -showRep:false").format(cwd,
214 enabled_testcases_str,
220 # Make sure to create the necessary output sub-folders for MTS
221 # and cleanup output files from previous run.
222 if os.path.exists(self.mts_result_csv_file):
223 os.remove(self.mts_result_csv_file)
225 if os.path.isdir(self.mts_stats_dir):
226 shutil.rmtree(self.mts_stats_dir)
227 os.makedirs(self.mts_stats_dir)
229 if os.path.isdir(self.mts_logs_dir):
230 shutil.rmtree(self.mts_logs_dir)
231 os.makedirs(self.mts_logs_dir)
234 "MTS statistics output dir: %s ", self.mts_stats_dir)
235 self.__logger.info("MTS logs output dir: %s ", self.mts_logs_dir)
237 # Launch MTS as a sub-process
238 # and save its standard output to a file
239 with open(self.result_file, 'w') as f_stdout:
240 self.__logger.info("Calling %s", cmd)
241 process = subprocess.Popen(
242 cmd, shell=True, stdout=subprocess.PIPE,
243 stderr=subprocess.STDOUT)
244 for line in iter(process.stdout.readline, b''):
246 sys.stdout.write(line.decode("utf-8"))
247 f_stdout.write(line.decode("utf-8"))
250 process.wait(timeout=max_duration)
251 except subprocess.TimeoutExpired:
254 "Killing MTS process after %d second(s).",
259 with open(self.result_file, 'r') as f_stdin:
260 self.__logger.debug("$ %s\n%s", cmd, f_stdin.read().rstrip())
261 return process.returncode
263 self.__logger.error("Missing mandatory arg for MTS. kwargs: %s",
267 def run(self, **kwargs):
270 It allows executing any Python method by calling execute().
272 It sets the following attributes required to push the results
279 It doesn't fulfill details when pushing the results to the DB.
282 kwargs: Arbitrary keyword arguments.
285 TestCase.EX_OK if execute() returns 0,
286 TestCase.EX_RUN_ERROR otherwise.
288 self.start_time = time.time()
289 exit_code = testcase.TestCase.EX_RUN_ERROR
292 nb_testcases = self.parse_xml_test_file(kwargs["test_file"])
293 # Do something only if there are some MTS test cases in the test
296 if self.execute(**kwargs) == 0:
297 exit_code = testcase.TestCase.EX_OK
300 except Exception: # pylint: disable=broad-except
301 self.__logger.exception(
302 "Cannot parse result file "
303 "$MTS_HOME/logs/testPlan.csv")
304 exit_code = testcase.TestCase.EX_RUN_ERROR
305 except Exception: # pylint: disable=broad-except
306 self.__logger.exception("%s FAILED", self.project_name)
307 self.stop_time = time.time()