#!/usr/bin/env python # coding: utf-8 # Copyright (c) 2020 Huawei Technologies Co., Ltd. # oec-hardware is licensed under the Mulan PSL v2. # You can use this software according to the terms and conditions of the Mulan PSL v2. # You may obtain a copy of Mulan PSL v2 at: # http://license.coscl.org.cn/MulanPSL2 # THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR # IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR # PURPOSE. # See the Mulan PSL v2 for more details. # Create: 2020-04-01 import os import sys import string import random import argparse from .test import Test from .env import CertEnv from .command import Command, CertCommandError from .commandUI import CommandUI from .log import Logger from .reboot import Reboot class Job(object): """ Test task management """ def __init__(self, args=None): """ Creates an instance of Job class. :param args: the job configuration, usually set by command line options and argument parsing :type args: :class:`argparse.Namespace` """ self.args = args or argparse.Namespace() self.test_factory = getattr(args, "test_factory", []) self.test_suite = [] self.job_id = ''.join(random.sample(string.ascii_letters + string.digits, 10)) self.ui = CommandUI() self.subtests_filter = getattr(args, "subtests_filter", None) self.test_parameters = None if "test_parameters" in self.args: self.test_parameters = {} for parameter_name, parameter_value in self.args.test_parameters: self.test_parameters[parameter_name] = parameter_value def discover(self, testname, subtests_filter=None): """ discover test :param testname: :param subtests_filter: :return: """ if not testname: print("testname not specified, discover test failed") return None filename = testname + ".py" dirpath = '' for (dirpath, dirs, files) in os.walk(CertEnv.testdirectoy): if filename in files: break pth = os.path.join(dirpath, filename) if not os.access(pth, os.R_OK): return None sys.path.insert(0, dirpath) try: module = __import__(testname, globals(), locals()) except Exception as e: print("Error: module import failed for %s" % testname) print(e) return None for thing in dir(module): test_class = getattr(module, thing) try: from types import ClassType as ct except ImportError: ct = type if isinstance(test_class, ct) and issubclass(test_class, Test): if "test" not in dir(test_class): continue if (subtests_filter and not subtests_filter in dir(test_class)): continue test = test_class() if "pri" not in dir(test): continue return test return None def create_test_suite(self, subtests_filter=None): """ Create test suites :param subtests_filter: :return: """ if self.test_suite: return self.test_suite = [] for test in self.test_factory: if test["run"]: testclass = self.discover(test["name"], subtests_filter) if testclass: testcase = dict() testcase["test"] = testclass testcase["name"] = test["name"] testcase["device"] = test["device"] testcase["status"] = "FAIL" self.test_suite.append(testcase) else: if not subtests_filter: test["status"] = "FAIL" print("not found %s" % test["name"]) if not len(self.test_suite): print("No test found") def check_test_depends(self): """ Install dependency packages :return: depending """ required_rpms = [] for tests in self.test_suite: for pkg in tests["test"].requirements: try: Command("rpm -q " + pkg).run_quiet() except CertCommandError: if not pkg in required_rpms: required_rpms.append(pkg) if len(required_rpms): print("Installing required packages: %s" % ", ".join(required_rpms)) try: cmd = Command("yum install -y " + " ".join(required_rpms)) cmd.echo() except CertCommandError as e: print(e) print("Fail to install required packages.") return False return True def _run_test(self, testcase, subtests_filter=None): """ Start a testing item :param testcase: :param subtests_filter: :return: """ name = testcase["name"] if testcase["device"].get_name(): name = testcase["name"] + "-" + testcase["device"].get_name() logname = name + ".log" reboot = None test = None logger = None try: test = testcase["test"] logger = Logger(logname, self.job_id, sys.stdout, sys.stderr) logger.start() if subtests_filter: return_code = getattr(test, subtests_filter)() else: print("---- start to run test %s ----" % name) args = argparse.Namespace(device=testcase["device"], logdir=logger.log.dir) test.setup(args) if test.reboot: reboot = Reboot(testcase["name"], self, test.rebootup) return_code = False if reboot.setup(): return_code = test.test() else: return_code = test.test() except Exception as e: print(e) return_code = False if reboot: reboot.clean() if not subtests_filter: test.teardown() logger.stop() print("") return return_code def run_tests(self, subtests_filter=None): """ Start testing :param subtests_filter: :return: """ if not len(self.test_suite): print("No test to run.") return self.test_suite.sort(key=lambda k: k["test"].pri) for testcase in self.test_suite: if self._run_test(testcase, subtests_filter): testcase["status"] = "PASS" else: testcase["status"] = "FAIL" def run(self): """ Test entrance :return: """ logger = Logger("job.log", self.job_id, sys.stdout, sys.stderr) logger.start() self.create_test_suite(self.subtests_filter) if not self.check_test_depends(): print("Required rpm package not installed, test stopped.") else: self.run_tests(self.subtests_filter) self.save_result() logger.stop() self.show_summary() def show_summary(self): """ Command line interface display summary :return: """ print("------------- Summary -------------") for test in self.test_factory: if test["run"]: name = test["name"] if test["device"].get_name(): name = test["name"] + "-" + test["device"].get_name() if test["status"] == "PASS": print(name.ljust(33) + "\033[0;32mPASS\033[0m") else: print(name.ljust(33) + "\033[0;31mFAIL\033[0m") print("") def save_result(self): """ Get test status :return: """ for test in self.test_factory: for testcase in self.test_suite: if test["name"] == testcase["name"] and test["device"].path == testcase["device"].path: test["status"] = testcase["status"]