job.py

#!/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"]