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 __init__(self, **kwargs):
44 super(MTSLauncher, self).__init__(**kwargs)
45 self.result_file = "{}/{}.log".format(self.res_dir, self.case_name)
46 # Location of the HTML report generated by MTS
47 self.mts_stats_dir = os.path.join(self.res_dir, 'mts_stats_report')
48 # Location of the log files generated by MTS for each test.
49 # Need to end path with a separator because of a bug in MTS.
50 self.mts_logs_dir = os.path.join(self.res_dir,
51 'mts_logs' + os.path.sep)
52 # The location of file named testPlan.csv
53 # that it always in $MTS_HOME/logs
54 self.mts_result_csv_file = self.mts_install_dir + os.path.sep
55 self.mts_result_csv_file += ("logs" + os.path.sep + "testPlan.csv")
63 def parse_results(self):
64 """Parse testPlan.csv containing the status of each testcase of the test file.
65 See sample file in `xtesting/samples/mts/output/testPlan.csv`
67 with open(self.mts_result_csv_file) as stream_:
68 self.__logger.info("Parsing file : %s", self.mts_result_csv_file)
69 reader = csv.reader(stream_, delimiter=';')
72 msg = prettytable.PrettyTable(
73 header_style='upper', padding_width=5,
74 field_names=['MTS test', 'MTS test case',
80 # If there's only one delimiter,
81 # it is the name of the <test> elt
84 _test_dict['parent'] = test_name
86 testcase_name = row[0].lstrip()
87 testcase_status = row[2]
89 if testcase_status == 'OK':
91 elif testcase_status == 'Failed':
93 elif testcase_status == '?':
95 _test_dict['status'] = testcase_status
96 _test_dict['name'] = testcase_name
100 _test_dict['status']])
102 _tests_data.append(_test_dict)
104 self.result = 100 * (
105 self.pass_tests / self.total_tests)
106 except ZeroDivisionError:
107 self.__logger.error("No test has been run")
108 self.__logger.info("MTS Test result:\n\n%s\n", msg.get_string())
110 self.details['description'] = "Execution of some MTS tests"
111 self.details['total_tests'] = self.total_tests
112 self.details['pass_tests'] = self.pass_tests
113 self.details['fail_tests'] = self.fail_tests
114 self.details['skip_tests'] = self.skip_tests
115 self.details['tests'] = _tests_data
117 def parse_xml_test_file(self, xml_test_file):
118 """Parse the XML file containing the test definition for MTS.
119 See sample file in `xtesting/samples/mts/test.xml`
123 "Parsing XML test file %s containing the MTS tests definitions.",
126 parser = etree.XMLParser(load_dtd=True, resolve_entities=True)
127 self.__logger.info("XML test file %s successfully parsed.",
129 root = etree.parse(xml_test_file, parser=parser)
130 # Need to look at all child nodes because there may be
131 # some <for> elt between <test> and <testcase> elt
132 self.testcases = root.xpath('//test//testcase/@name')
133 nb_testcases = len(self.testcases)
134 if nb_testcases == 0:
135 self.__logger.warning("Found no MTS testcase !")
136 elif nb_testcases == 1:
137 self.__logger.info("Found only one MTS testcase: %s",
140 self.__logger.info("Found %d MTS testcases :", nb_testcases)
141 for mts_testcase in self.testcases:
142 self.__logger.info(" - %s", mts_testcase)
143 except etree.XMLSyntaxError as xml_err:
144 self.__logger.error("Error while parsing XML test file: %s",
148 def check_enabled_mts_test_cases(self, enabled_testcases):
149 """Make sure that all required MTS test cases exist
150 in the XML test file.
152 if len(enabled_testcases) > 0:
153 # Verify if the MTS test case exists in the whole list of test
154 # cases declared in the test XML file
155 for enabled_testcase in enabled_testcases:
156 if enabled_testcase not in self.testcases:
158 "The required MTS testcase named `%s` does not exist"
159 " !", enabled_testcase)
163 def execute(self, **kwargs): # pylint: disable=too-many-locals
164 """Execute the cmd passed as arg
167 kwargs: Arbitrary keyword arguments.
174 console = kwargs["console"] if "console" in kwargs else False
175 # Read specific parameters for MTS
176 test_file = kwargs["test_file"]
178 "log_level"] if "log_level" in kwargs else "INFO"
180 # For some MTS tests, we need to force stop after N sec
181 max_duration = kwargs[
182 "max_duration"] if "max_duration" in kwargs else None
183 store_method = kwargs[
184 "store_method"] if "store_method" in kwargs else "FILE"
185 # Must use the $HOME_MTS/bin as current working dir
186 cwd = self.mts_install_dir + os.path.sep + "bin"
188 # Get the list of enabled MTS testcases, if any
189 enabled_testcases = kwargs[
190 "testcases"] if "testcases" in kwargs else []
191 enabled_testcases_str = ''
192 if len(enabled_testcases) > 0:
193 enabled_testcases_str = ' '.join(enabled_testcases)
194 check_ok = self.check_enabled_mts_test_cases(enabled_testcases)
198 # Build command line to launch for MTS
199 cmd = ("cd {} && ./startCmd.sh {} {} -sequential -levelLog:{}"
201 " -config:stats.REPORT_DIRECTORY+{}"
202 " -config:logs.STORAGE_DIRECTORY+{}"
204 " -showRep:false").format(cwd,
206 enabled_testcases_str,
212 # Make sure to create the necessary output sub-folders for MTS
213 # and cleanup output files from previous run.
214 if os.path.exists(self.mts_result_csv_file):
215 os.remove(self.mts_result_csv_file)
217 if os.path.isdir(self.mts_stats_dir):
218 shutil.rmtree(self.mts_stats_dir)
219 os.makedirs(self.mts_stats_dir)
221 if os.path.isdir(self.mts_logs_dir):
222 shutil.rmtree(self.mts_logs_dir)
223 os.makedirs(self.mts_logs_dir)
226 "MTS statistics output dir: %s ", self.mts_stats_dir)
227 self.__logger.info("MTS logs output dir: %s ", self.mts_logs_dir)
229 # Launch MTS as a sub-process
230 # and save its standard output to a file
231 with open(self.result_file, 'w') as f_stdout:
232 self.__logger.info("Calling %s", cmd)
233 process = subprocess.Popen(
234 cmd, shell=True, stdout=subprocess.PIPE,
235 stderr=subprocess.STDOUT)
236 for line in iter(process.stdout.readline, b''):
238 sys.stdout.write(line.decode("utf-8"))
239 f_stdout.write(line.decode("utf-8"))
242 process.wait(timeout=max_duration)
243 except subprocess.TimeoutExpired:
246 "Killing MTS process after %d second(s).",
251 with open(self.result_file, 'r') as f_stdin:
252 self.__logger.debug("$ %s\n%s", cmd, f_stdin.read().rstrip())
253 return process.returncode
255 self.__logger.error("Missing mandatory arg for MTS. kwargs: %s",
259 def run(self, **kwargs):
262 It allows executing any Python method by calling execute().
264 It sets the following attributes required to push the results
271 It doesn't fulfill details when pushing the results to the DB.
274 kwargs: Arbitrary keyword arguments.
277 TestCase.EX_OK if execute() returns 0,
278 TestCase.EX_RUN_ERROR otherwise.
280 self.start_time = time.time()
281 exit_code = testcase.TestCase.EX_RUN_ERROR
284 nb_testcases = self.parse_xml_test_file(kwargs["test_file"])
285 # Do something only if there are some MTS test cases in the test
288 if self.execute(**kwargs) == 0:
289 exit_code = testcase.TestCase.EX_OK
292 except Exception: # pylint: disable=broad-except
293 self.__logger.exception(
294 "Cannot parse result file "
295 "$MTS_HOME/logs/testPlan.csv")
296 exit_code = testcase.TestCase.EX_RUN_ERROR
297 except Exception: # pylint: disable=broad-except
298 self.__logger.exception("%s FAILED", self.project_name)
299 self.stop_time = time.time()