diff options
author | 2023-04-14 15:19:17 +0200 | |
---|---|---|
committer | 2023-04-17 15:38:04 +0200 | |
commit | b9472413a7c759bc2d2f1b9a0ab2c229fd36f547 (patch) | |
tree | a6829319b89fc2430ca33f90852cfde77aaf2917 /xtesting/core | |
parent | f88f3076d913394a75871fcec65de0e6cd497289 (diff) |
Remove MTS from Xtesting
The MTS drivers is unmaintained and they are lots of security issues
in its dependencies.
Change-Id: I397e22c8868204b8d52a8ac35906278b5e530961
Signed-off-by: Cédric Ollivier <cedric.ollivier@orange.com>
(cherry picked from commit 781fc313c1c59018edb892ff0423963191781592)
Diffstat (limited to 'xtesting/core')
-rw-r--r-- | xtesting/core/mts.py | 308 |
1 files changed, 0 insertions, 308 deletions
diff --git a/xtesting/core/mts.py b/xtesting/core/mts.py deleted file mode 100644 index d6865a61..00000000 --- a/xtesting/core/mts.py +++ /dev/null @@ -1,308 +0,0 @@ -#!/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 - -"""Define the parent classes of all Xtesting Features. - -Feature is considered as TestCase offered by Third-party. It offers -helpers to run any python method or any bash command. -""" - -import csv -import logging -import os -import shutil -import subprocess -import sys -import time - -from lxml import etree -import prettytable -import six - -from xtesting.core import testcase - - -__author__ = ("Vincent Mahe <v.mahe@orange.com>, " - "Cedric Ollivier <cedric.ollivier@orange.com>") - - -class MTSLauncher(testcase.TestCase): - """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(MTSLauncher, self).__init__(**kwargs) - self.result_file = "{}/{}.log".format(self.res_dir, self.case_name) - # 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) 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 - """Execute the cmd passed as arg - - Args: - kwargs: Arbitrary keyword arguments. - - Returns: - 0 if cmd returns 0, - -1 otherwise. - """ - try: - console = kwargs["console"] if "console" in kwargs else False - # Read specific parameters for MTS - test_file = kwargs["test_file"] - log_level = kwargs[ - "log_level"] if "log_level" in kwargs else "INFO" - - # For some MTS tests, we need to force stop after N sec - max_duration = kwargs[ - "max_duration"] if "max_duration" in kwargs else None - 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 -2 - - # Build command line to launch for MTS - cmd = ("cd {} && ./startCmd.sh {} {} -sequential -levelLog:{}" - " -storageLog:{}" - " -config:stats.REPORT_DIRECTORY+{}" - " -config:logs.STORAGE_DIRECTORY+{}" - " -genReport:true" - " -showRep:false").format(cwd, - test_file, - enabled_testcases_str, - log_level, - store_method, - self.mts_stats_dir, - self.mts_logs_dir) - - # 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) - - # Launch MTS as a sub-process - # and save its standard output to a file - with open(self.result_file, 'w') as f_stdout: - self.__logger.info("Calling %s", cmd) - process = subprocess.Popen( - cmd, shell=True, stdout=subprocess.PIPE, - stderr=subprocess.STDOUT) - for line in iter(process.stdout.readline, b''): - if console: - sys.stdout.write(line.decode("utf-8")) - f_stdout.write(line.decode("utf-8")) - if six.PY3: - try: - process.wait(timeout=max_duration) - except subprocess.TimeoutExpired: - process.kill() - self.__logger.info( - "Killing MTS process after %d second(s).", - max_duration) - return 3 - else: - process.wait() - with open(self.result_file, 'r') as f_stdin: - self.__logger.debug("$ %s\n%s", cmd, f_stdin.read().rstrip()) - return process.returncode - except KeyError: - self.__logger.error("Missing mandatory arg for MTS. kwargs: %s", - kwargs) - return -1 - - def run(self, **kwargs): - """Run the feature. - - It allows executing any Python method by calling execute(). - - It sets the following attributes required to push the results - to DB: - - * result, - * start_time, - * stop_time. - - It doesn't fulfill details when pushing the results to the DB. - - Args: - kwargs: Arbitrary keyword arguments. - - Returns: - TestCase.EX_OK if execute() returns 0, - TestCase.EX_RUN_ERROR otherwise. - """ - 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 |