path: root/meta/lib/oeqa/core/runner.py

# Copyright (C) 2016 Intel Corporation
# Released under the MIT license (see COPYING.MIT)

import os
import time
import unittest
import logging
import re
import json
import pathlib

from unittest import TextTestResult as _TestResult
from unittest import TextTestRunner as _TestRunner

class OEStreamLogger(object):
    def __init__(self, logger):
        self.logger = logger
        self.buffer = ""

    def write(self, msg):
        if len(msg) > 1 and msg[0] != '\n':
            if '...' in msg:
                self.buffer += msg
            elif self.buffer:
                self.buffer += msg
                self.logger.log(logging.INFO, self.buffer)
                self.buffer = ""
            else:
                self.logger.log(logging.INFO, msg)

    def flush(self):
        for handler in self.logger.handlers:
            handler.flush()

class OETestResult(_TestResult):
    def __init__(self, tc, *args, **kwargs):
        super(OETestResult, self).__init__(*args, **kwargs)

        self.successes = []
        self.starttime = {}
        self.endtime = {}
        self.progressinfo = {}

        # Inject into tc so that TestDepends decorator can see results
        tc.results = self

        self.tc = tc

        self.result_types = ['failures', 'errors', 'skipped', 'expectedFailures', 'successes']
        self.result_desc = ['FAILED', 'ERROR', 'SKIPPED', 'EXPECTEDFAIL', 'PASSED']

    def startTest(self, test):
        # May have been set by concurrencytest
        if test.id() not in self.starttime:
            self.starttime[test.id()] = time.time()
        super(OETestResult, self).startTest(test)

    def stopTest(self, test):
        self.endtime[test.id()] = time.time()
        super(OETestResult, self).stopTest(test)
        if test.id() in self.progressinfo:
            self.tc.logger.info(self.progressinfo[test.id()])

        # Print the errors/failures early to aid/speed debugging, its a pain
        # to wait until selftest finishes to see them.
        for t in ['failures', 'errors', 'skipped', 'expectedFailures']:
            for (scase, msg) in getattr(self, t):
                if test.id() == scase.id():
                    self.tc.logger.info(str(msg))
                    break

    def logSummary(self, component, context_msg=''):
        elapsed_time = self.tc._run_end_time - self.tc._run_start_time
        self.tc.logger.info("SUMMARY:")
        self.tc.logger.info("%s (%s) - Ran %d test%s in %.3fs" % (component,
            context_msg, self.testsRun, self.testsRun != 1 and "s" or "",
            elapsed_time))

        if self.wasSuccessful():
            msg = "%s - OK - All required tests passed" % component
        else:
            msg = "%s - FAIL - Required tests failed" % component
        skipped = len(self.skipped)
        if skipped: 
            msg += " (skipped=%d)" % skipped
        self.tc.logger.info(msg)

    def _isTestResultContainTestCaseWithResultTypeProvided(self, case, type):
        found = False

        for (scase, msg) in getattr(self, type):
            if case.id() == scase.id():
                found = True
                break
            scase_str = str(scase.id())

            # When fails at module or class level the class name is passed as string
            # so figure out to see if match
            m = re.search("^setUpModule \((?P<module_name>.*)\)$", scase_str)
            if m:
                if case.__class__.__module__ == m.group('module_name'):
                    found = True
                    break

            m = re.search("^setUpClass \((?P<class_name>.*)\)$", scase_str)
            if m:
                class_name = "%s.%s" % (case.__class__.__module__,
                        case.__class__.__name__)

                if class_name == m.group('class_name'):
                    found = True
                    break

        if found:
            return (found, msg)

        return (found, None)

    def addSuccess(self, test):
        #Added so we can keep track of successes too
        self.successes.append((test, None))
        super(OETestResult, self).addSuccess(test)

    def logDetails(self):
        self.tc.logger.info("RESULTS:")
        for case_name in self.tc._registry['cases']:
            case = self.tc._registry['cases'][case_name]

            found = False
            desc = None
            for idx, name in enumerate(self.result_types):
                (found, msg) = self._isTestResultContainTestCaseWithResultTypeProvided(case, self.result_types[idx])
                if found:
                    desc = self.result_desc[idx]
                    break

            oeid = -1
            if hasattr(case, 'decorators'):
                for d in case.decorators:
                    if hasattr(d, 'oeid'):
                        oeid = d.oeid

            t = ""
            if case.id() in self.starttime and case.id() in self.endtime:
                t = " (" + "{0:.2f}".format(self.endtime[case.id()] - self.starttime[case.id()]) + "s)"

            if found:
                self.tc.logger.info("RESULTS - %s - Testcase %s: %s%s" % (case.id(),
                    oeid, desc, t))
            else:
                self.tc.logger.info("RESULTS - %s - Testcase %s: %s%s" % (case.id(),
                    oeid, 'UNKNOWN', t))

    def _get_testcase_result_and_testmessage_dict(self):
        testcase_result_dict = {}
        testcase_testmessage_dict = {}
        for case_name in self.tc._registry['cases']:
            case = self.tc._registry['cases'][case_name]

            found = False
            desc = None
            test_msg = ''
            for idx, name in enumerate(self.result_types):
                (found, msg) = self._isTestResultContainTestCaseWithResultTypeProvided(case, self.result_types[idx])
                if found:
                    desc = self.result_desc[idx]
                    test_msg = msg
                    break

            if found:
                testcase_result_dict[case.id()] = desc
                testcase_testmessage_dict[case.id()] = test_msg
            else:
                testcase_result_dict[case.id()] = "UNKNOWN"
        return testcase_result_dict, testcase_testmessage_dict

    def logDetailsInJson(self, file_dir):
        (testcase_result_dict, testcase_testmessage_dict) = self._get_testcase_result_and_testmessage_dict()
        if len(testcase_result_dict) > 0 and len(testcase_testmessage_dict) > 0:
            jsontresulthelper = OEJSONTestResultHelper(testcase_result_dict, testcase_testmessage_dict)
            jsontresulthelper.write_json_testresult_files(file_dir)
            jsontresulthelper.write_testcase_log_files(os.path.join(file_dir, 'logs'))

class OEListTestsResult(object):
    def wasSuccessful(self):
        return True

class OETestRunner(_TestRunner):
    streamLoggerClass = OEStreamLogger

    def __init__(self, tc, *args, **kwargs):
        kwargs['stream'] = self.streamLoggerClass(tc.logger)
        super(OETestRunner, self).__init__(*args, **kwargs)
        self.tc = tc
        self.resultclass = OETestResult

    def _makeResult(self):
        return self.resultclass(self.tc, self.stream, self.descriptions,
                self.verbosity)

    def _walk_suite(self, suite, func):
        for obj in suite:
            if isinstance(obj, unittest.suite.TestSuite):
                if len(obj._tests):
                    self._walk_suite(obj, func)
            elif isinstance(obj, unittest.case.TestCase):
                func(self.tc.logger, obj)
                self._walked_cases = self._walked_cases + 1

    def _list_tests_name(self, suite):
        from oeqa.core.decorator.oeid import OETestID
        from oeqa.core.decorator.oetag import OETestTag

        self._walked_cases = 0

        def _list_cases_without_id(logger, case):

            found_id = False
            if hasattr(case, 'decorators'):
                for d in case.decorators:
                    if isinstance(d, OETestID):
                        found_id = True

            if not found_id:
                logger.info('oeid missing for %s' % case.id())

        def _list_cases(logger, case):
            oeid = None
            oetag = None

            if hasattr(case, 'decorators'):
                for d in case.decorators:
                    if isinstance(d, OETestID):
                        oeid = d.oeid
                    elif isinstance(d, OETestTag):
                        oetag = d.oetag

            logger.info("%s\t%s\t\t%s" % (oeid, oetag, case.id()))

        self.tc.logger.info("Listing test cases that don't have oeid ...")
        self._walk_suite(suite, _list_cases_without_id)
        self.tc.logger.info("-" * 80)

        self.tc.logger.info("Listing all available tests:")
        self._walked_cases = 0
        self.tc.logger.info("id\ttag\t\ttest")
        self.tc.logger.info("-" * 80)
        self._walk_suite(suite, _list_cases)
        self.tc.logger.info("-" * 80)
        self.tc.logger.info("Total found:\t%s" % self._walked_cases)

    def _list_tests_class(self, suite):
        self._walked_cases = 0

        curr = {}
        def _list_classes(logger, case):
            if not 'module' in curr or curr['module'] != case.__module__:
                curr['module'] = case.__module__
                logger.info(curr['module'])

            if not 'class' in curr  or curr['class'] != \
                    case.__class__.__name__:
                curr['class'] = case.__class__.__name__
                logger.info(" -- %s" % curr['class'])

            logger.info(" -- -- %s" % case._testMethodName)

        self.tc.logger.info("Listing all available test classes:")
        self._walk_suite(suite, _list_classes)

    def _list_tests_module(self, suite):
        self._walked_cases = 0

        listed = []
        def _list_modules(logger, case):
            if not case.__module__ in listed:
                if case.__module__.startswith('_'):
                    logger.info("%s (hidden)" % case.__module__)
                else:
                    logger.info(case.__module__)
                listed.append(case.__module__)

        self.tc.logger.info("Listing all available test modules:")
        self._walk_suite(suite, _list_modules)

    def list_tests(self, suite, display_type):
        if display_type == 'name':
            self._list_tests_name(suite)
        elif display_type == 'class':
            self._list_tests_class(suite)
        elif display_type == 'module':
            self._list_tests_module(suite)

        return OEListTestsResult()

class OEJSONTestResultHelper(object):
    def __init__(self, testcase_result_dict, testcase_log_dict):
        self.testcase_result_dict = testcase_result_dict
        self.testcase_log_dict = testcase_log_dict

    def get_testcase_list(self):
        return self.testcase_result_dict.keys()

    def get_testsuite_from_testcase(self, testcase):
        testsuite = testcase[0:testcase.rfind(".")]
        return testsuite

    def get_testmodule_from_testsuite(self, testsuite):
        testmodule = testsuite[0:testsuite.find(".")]
        return testmodule

    def get_testsuite_testcase_dictionary(self):
        testsuite_testcase_dict = {}
        for testcase in self.get_testcase_list():
            testsuite = self.get_testsuite_from_testcase(testcase)
            if testsuite in testsuite_testcase_dict:
                testsuite_testcase_dict[testsuite].append(testcase)
            else:
                testsuite_testcase_dict[testsuite] = [testcase]
        return testsuite_testcase_dict

    def get_testmodule_testsuite_dictionary(self, testsuite_testcase_dict):
        testsuite_list = testsuite_testcase_dict.keys()
        testmodule_testsuite_dict = {}
        for testsuite in testsuite_list:
            testmodule = self.get_testmodule_from_testsuite(testsuite)
            if testmodule in testmodule_testsuite_dict:
                testmodule_testsuite_dict[testmodule].append(testsuite)
            else:
                testmodule_testsuite_dict[testmodule] = [testsuite]
        return testmodule_testsuite_dict

    def _get_testcase_result(self, testcase, testcase_status_dict):
        if testcase in testcase_status_dict:
            return testcase_status_dict[testcase]
        return ""

    def _create_testcase_testresult_object(self, testcase_list, testcase_result_dict):
        testcase_dict = {}
        for testcase in sorted(testcase_list):
            result = self._get_testcase_result(testcase, testcase_result_dict)
            testcase_dict[testcase] = {"testresult": result}
        return testcase_dict

    def _create_json_testsuite_string(self, testsuite_list, testsuite_testcase_dict, testcase_result_dict):
        testsuite_object = {'testsuite': {}}
        testsuite_dict = testsuite_object['testsuite']
        for testsuite in sorted(testsuite_list):
            testsuite_dict[testsuite] = {'testcase': {}}
            testsuite_dict[testsuite]['testcase'] = self._create_testcase_testresult_object(
                testsuite_testcase_dict[testsuite],
                testcase_result_dict)
        return json.dumps(testsuite_object, sort_keys=True, indent=4)

    def write_json_testresult_files(self, write_dir):
        if not os.path.exists(write_dir):
            pathlib.Path(write_dir).mkdir(parents=True, exist_ok=True)
        testsuite_testcase_dict = self.get_testsuite_testcase_dictionary()
        testmodule_testsuite_dict = self.get_testmodule_testsuite_dictionary(testsuite_testcase_dict)
        for testmodule in testmodule_testsuite_dict.keys():
            testsuite_list = testmodule_testsuite_dict[testmodule]
            json_testsuite = self._create_json_testsuite_string(testsuite_list, testsuite_testcase_dict,
                                                                self.testcase_result_dict)
            file_name = '%s.json' % testmodule
            file_path = os.path.join(write_dir, file_name)
            with open(file_path, 'w') as the_file:
                the_file.write(json_testsuite)

    def write_testcase_log_files(self, write_dir):
        if not os.path.exists(write_dir):
            pathlib.Path(write_dir).mkdir(parents=True, exist_ok=True)
        for testcase in self.testcase_log_dict.keys():
            test_log = self.testcase_log_dict[testcase]
            if test_log is not None:
                file_name = '%s.log' % testcase
                file_path = os.path.join(write_dir, file_name)
                with open(file_path, 'w') as the_file:
                    the_file.write(test_log)