#!/usr/bin/env python
# Copyright (c) 2020 Orange and others.
#
# All rights reserved. This program and the accompanying materials
# are made available under the terms of the Apache License, Version 2.0
# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
# pylint: disable=too-many-instance-attributes
"""Implement a Xtesting driver to run mts suite."""
import csv
import logging
import os
import shutil
import time
from lxml import etree
import prettytable
from xtesting.core import feature
from xtesting.core import testcase
__author__ = ("Vincent Mahe <v.mahe@orange.com>, "
"Cedric Ollivier <cedric.ollivier@orange.com>")
class MTSLauncher(feature.BashFeature):
"""Class designed to run MTS tests."""
__logger = logging.getLogger(__name__)
mts_install_dir = "/opt/mts"
def check_requirements(self):
"""Check if startCmd.sh is in /opt/mts/bin"""
if not os.path.exists(
os.path.join(self.mts_install_dir, 'bin/startCmd.sh')):
self.__logger.warning(
"mts is not available for arm for the time being")
self.is_skipped = True
def __init__(self, **kwargs):
super().__init__(**kwargs)
# Location of the HTML report generated by MTS
self.mts_stats_dir = os.path.join(self.res_dir, 'mts_stats_report')
# Location of the log files generated by MTS for each test.
# Need to end path with a separator because of a bug in MTS.
self.mts_logs_dir = os.path.join(self.res_dir,
'mts_logs' + os.path.sep)
# The location of file named testPlan.csv
# that it always in $MTS_HOME/logs
self.mts_result_csv_file = self.mts_install_dir + os.path.sep
self.mts_result_csv_file += ("logs" + os.path.sep + "testPlan.csv")
self.total_tests = 0
self.pass_tests = 0
self.fail_tests = 0
self.skip_tests = 0
self.response = None
self.testcases = []
def parse_results(self):
"""Parse testPlan.csv containing the status of each testcase of the test file.
See sample file in `xtesting/samples/mts/output/testPlan.csv`
"""
with open(self.mts_result_csv_file, encoding='utf-8') as stream_:
self.__logger.info("Parsing file : %s", self.mts_result_csv_file)
reader = csv.reader(stream_, delimiter=';')
rownum = 0
_tests_data = []
msg = prettytable.PrettyTable(
header_style='upper', padding_width=5,
field_names=['MTS test', 'MTS test case',
'status'])
for row in reader:
_test_dict = {}
nb_values = len(row)
if rownum > 0:
# If there's only one delimiter,
# it is the name of the <test> elt
if nb_values == 2:
test_name = row[0]
_test_dict['parent'] = test_name
elif nb_values == 3:
testcase_name = row[0].lstrip()
testcase_status = row[2]
self.total_tests += 1
if testcase_status == 'OK':
self.pass_tests += 1
elif testcase_status == 'Failed':
self.fail_tests += 1
elif testcase_status == '?':
self.skip_tests += 1
_test_dict['status'] = testcase_status
_test_dict['name'] = testcase_name
msg.add_row(
[test_name,
_test_dict['name'],
_test_dict['status']])
rownum += 1
_tests_data.append(_test_dict)
try:
self.result = 100 * (
self.pass_tests / self.total_tests)
except ZeroDivisionError:
self.__logger.error("No test has been run")
self.__logger.info("MTS Test result:\n\n%s\n", msg.get_string())
self.details = {}
self.details['description'] = "Execution of some MTS tests"
self.details['total_tests'] = self.total_tests
self.details['pass_tests'] = self.pass_tests
self.details['fail_tests'] = self.fail_tests
self.details['skip_tests'] = self.skip_tests
self.details['tests'] = _tests_data
def parse_xml_test_file(self, xml_test_file):
"""Parse the XML file containing the test definition for MTS.
See sample file in `xtesting/samples/mts/test.xml`
"""
nb_testcases = -1
self.__logger.info(
"Parsing XML test file %s containing the MTS tests definitions.",
xml_test_file)
try:
parser = etree.XMLParser(load_dtd=True, resolve_entities=True)
self.__logger.info("XML test file %s successfully parsed.",
xml_test_file)
root = etree.parse(xml_test_file, parser=parser)
# Need to look at all child nodes because there may be
# some <for> elt between <test> and <testcase> elt
self.testcases = root.xpath('//test//testcase/@name')
nb_testcases = len(self.testcases)
if nb_testcases == 0:
self.__logger.warning("Found no MTS testcase !")
elif nb_testcases == 1:
self.__logger.info("Found only one MTS testcase: %s",
self.testcases[0])
else:
self.__logger.info("Found %d MTS testcases :", nb_testcases)
for mts_testcase in self.testcases:
self.__logger.info(" - %s", mts_testcase)
except etree.XMLSyntaxError as xml_err:
self.__logger.error("Error while parsing XML test file: %s",
str(xml_err))
return nb_testcases
def check_enabled_mts_test_cases(self, enabled_testcases):
"""Make sure that all required MTS test cases exist
in the XML test file.
"""
if enabled_testcases:
# Verify if the MTS test case exists in the whole list of test
# cases declared in the test XML file
for enabled_testcase in enabled_testcases:
if enabled_testcase not in self.testcases:
self.__logger.error(
"The required MTS testcase named `%s` does not exist"
" !", enabled_testcase)
return False
return True
def execute(self, **kwargs): # pylint: disable=too-many-locals
try:
# Read specific parameters for MTS
test_file = kwargs["test_file"]
log_level = kwargs[
"log_level"] if "log_level" in kwargs else "INFO"
store_method = kwargs[
"store_method"] if "store_method" in kwargs else "FILE"
# Must use the $HOME_MTS/bin as current working dir
cwd = self.mts_install_dir + os.path.sep + "bin"
# Get the list of enabled MTS testcases, if any
enabled_testcases = kwargs[
"testcases"] if "testcases" in kwargs else []
enabled_testcases_str = ''
if enabled_testcases:
enabled_testcases_str = ' '.join(enabled_testcases)
check_ok = self.check_enabled_mts_test_cases(enabled_testcases)
if not check_ok:
return -3
# Build command line to launch for MTS
cmd = (f"cd {cwd} && ./startCmd.sh {test_file} "
f"{enabled_testcases_str} -sequential -levelLog:{log_level}"
f" -storageLog:{store_method}"
f" -config:stats.REPORT_DIRECTORY+{self.mts_stats_dir}"
f" -config:logs.STORAGE_DIRECTORY+{self.mts_logs_dir}"
" -genReport:true"
" -showRep:false")
# Make sure to create the necessary output sub-folders for MTS
# and cleanup output files from previous run.
if os.path.exists(self.mts_result_csv_file):
os.remove(self.mts_result_csv_file)
if os.path.isdir(self.mts_stats_dir):
shutil.rmtree(self.mts_stats_dir)
os.makedirs(self.mts_stats_dir)
if os.path.isdir(self.mts_logs_dir):
shutil.rmtree(self.mts_logs_dir)
os.makedirs(self.mts_logs_dir)
self.__logger.info(
"MTS statistics output dir: %s ", self.mts_stats_dir)
self.__logger.info("MTS logs output dir: %s ", self.mts_logs_dir)
kwargs.pop("cmd", None)
return super().execute(cmd=cmd, **kwargs)
except KeyError:
self.__logger.error("Missing mandatory arg for MTS. kwargs: %s",
kwargs)
return -1
def run(self, **kwargs):
"""Runs the MTS suite"""
self.start_time = time.time()
exit_code = testcase.TestCase.EX_RUN_ERROR
self.result = 0
try:
nb_testcases = self.parse_xml_test_file(kwargs["test_file"])
# Do something only if there are some MTS test cases in the test
# file
if nb_testcases > 0:
if self.execute(**kwargs) == 0:
exit_code = testcase.TestCase.EX_OK
try:
self.parse_results()
except Exception: # pylint: disable=broad-except
self.__logger.exception(
"Cannot parse result file "
"$MTS_HOME/logs/testPlan.csv")
exit_code = testcase.TestCase.EX_RUN_ERROR
except Exception: # pylint: disable=broad-except
self.__logger.exception("%s FAILED", self.project_name)
self.stop_time = time.time()
return exit_code