esp-idf/tools/ci/python_packages/tiny_test_fw/TinyFW.py

# Copyright 2015-2017 Espressif Systems (Shanghai) PTE LTD
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http:#www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

""" Interface for test cases. """
import json
import logging
import os
import time
import traceback
import functools
import socket
from datetime import datetime

import junit_xml

from . import Env
from . import DUT
from . import App
from . import Utility


class DefaultEnvConfig(object):
    """
    default test configs. There're 3 places to set configs, priority is (high -> low):

    1. overwrite set by caller of test method
    2. values set by test_method decorator
    3. default env config get from this class
    """
    DEFAULT_CONFIG = {
        "app": App.BaseApp,
        "dut": DUT.BaseDUT,
        "env_tag": "default",
        "env_config_file": None,
        "test_suite_name": None,
    }

    @classmethod
    def set_default_config(cls, **kwargs):
        """
        :param kwargs: configs need to be updated
        :return: None
        """
        cls.DEFAULT_CONFIG.update(kwargs)

    @classmethod
    def get_default_config(cls):
        """
        :return: current default config
        """
        return cls.DEFAULT_CONFIG.copy()


set_default_config = DefaultEnvConfig.set_default_config
get_default_config = DefaultEnvConfig.get_default_config

MANDATORY_INFO = {
    "execution_time": 1,
    "env_tag": "default",
    "category": "function",
    "ignore": False,
}


class JunitReport(object):
    # wrapper for junit test report
    # TODO: JunitReport methods are not thread safe (although not likely to be used this way).

    JUNIT_FILE_NAME = "XUNIT_RESULT.xml"
    JUNIT_DEFAULT_TEST_SUITE = "test-suite"
    JUNIT_TEST_SUITE = junit_xml.TestSuite(JUNIT_DEFAULT_TEST_SUITE,
                                           hostname=socket.gethostname(),
                                           timestamp=datetime.utcnow().isoformat())
    JUNIT_CURRENT_TEST_CASE = None
    _TEST_CASE_CREATED_TS = 0

    @classmethod
    def output_report(cls, junit_file_path):
        """ Output current test result to file. """
        with open(os.path.join(junit_file_path, cls.JUNIT_FILE_NAME), "w") as f:
            cls.JUNIT_TEST_SUITE.to_file(f, [cls.JUNIT_TEST_SUITE], prettyprint=False)

    @classmethod
    def get_current_test_case(cls):
        """
        By default, the test framework will handle junit test report automatically.
        While some test case might want to update some info to test report.
        They can use this method to get current test case created by test framework.

        :return: current junit test case instance created by ``JunitTestReport.create_test_case``
        """
        return cls.JUNIT_CURRENT_TEST_CASE

    @classmethod
    def test_case_finish(cls, test_case):
        """
        Append the test case to test suite so it can be output to file.
        Execution time will be automatically updated (compared to ``create_test_case``).
        """
        test_case.elapsed_sec = time.time() - cls._TEST_CASE_CREATED_TS
        cls.JUNIT_TEST_SUITE.test_cases.append(test_case)

    @classmethod
    def create_test_case(cls, name):
        """
        Extend ``junit_xml.TestCase`` with:

        1. save create test case so it can be get by ``get_current_test_case``
        2. log create timestamp, so ``elapsed_sec`` can be auto updated in ``test_case_finish``.

        :param name: test case name
        :return: instance of ``junit_xml.TestCase``
        """
        # set stdout to empty string, so we can always append string to stdout.
        # It won't affect output logic. If stdout is empty, it won't be put to report.
        test_case = junit_xml.TestCase(name, stdout="")
        cls.JUNIT_CURRENT_TEST_CASE = test_case
        cls._TEST_CASE_CREATED_TS = time.time()
        return test_case

    @classmethod
    def update_performance(cls, performance_items):
        """
        Update performance results to ``stdout`` of current test case.

        :param performance_items: a list of performance items. each performance item is a key-value pair.
        """
        assert cls.JUNIT_CURRENT_TEST_CASE

        for item in performance_items:
            cls.JUNIT_CURRENT_TEST_CASE.stdout += "[{}]: {}\n".format(item[0], item[1])


def test_method(**kwargs):
    """
    decorator for test case function.
    The following keyword arguments are pre-defined.
    Any other keyword arguments will be regarded as filter for the test case,
    able to access them by ``case_info`` attribute of test method.

    :keyword app: class for test app. see :doc:`App <App>` for details
    :keyword dut: class for current dut. see :doc:`DUT <DUT>` for details
    :keyword env_tag: name for test environment, used to select configs from config file
    :keyword env_config_file: test env config file. usually will not set this keyword when define case
    :keyword test_suite_name: test suite name, used for generating log folder name and adding xunit format test result.
                              usually will not set this keyword when define case
    :keyword junit_report_by_case: By default the test fw will handle junit report generation.
                                   In some cases, one test function might test many test cases.
                                   If this flag is set, test case can update junit report by its own.
    """

    def test(test_func):

        case_info = MANDATORY_INFO.copy()
        case_info["name"] = case_info["ID"] = test_func.__name__
        case_info["junit_report_by_case"] = False

        def _filter_ci_target(target, ci_target):
            if not ci_target:
                return target
            if isinstance(target, str):
                if isinstance(ci_target, str) and target == ci_target:
                    return ci_target
            else:
                if isinstance(ci_target, str) and ci_target in target:
                    return ci_target
                elif isinstance(ci_target, list) and set(ci_target).issubset(set(target)):
                    return ci_target
            raise ValueError('ci_target must be a subset of target')

        if os.getenv('CI_JOB_NAME') and 'ci_target' in kwargs:
            kwargs['target'] = _filter_ci_target(kwargs['target'], kwargs['ci_target'])
        case_info.update(kwargs)

        @functools.wraps(test_func)
        def handle_test(extra_data=None, **overwrite):
            """
            create env, run test and record test results

            :param extra_data: extra data that runner or main passed to test case
            :param overwrite: args that runner or main want to overwrite
            :return: None
            """
            # create env instance
            env_config = DefaultEnvConfig.get_default_config()
            for key in kwargs:
                if key in env_config:
                    env_config[key] = kwargs[key]

            # Runner.py should overwrite target with the current target.
            env_config.update(overwrite)

            # if target not in the default_config or the overwrite, then take it from kwargs passed from the decorator
            target = env_config['target'] if 'target' in env_config else kwargs['target']

            # This code block is used to do run local test script target set check
            if not os.getenv('CI_JOB_NAME'):
                idf_target = 'ESP32'  # default if sdkconfig not found or not readable
                expected_json_path = os.path.join('build', 'config', 'sdkconfig.json')
                if os.path.exists(expected_json_path):
                    sdkconfig = json.load(open(expected_json_path))
                    try:
                        idf_target = sdkconfig['IDF_TARGET'].upper()
                    except KeyError:
                        pass
                    else:
                        logging.info('IDF_TARGET: {}'.format(idf_target))
                else:
                    logging.warning('{} not found. IDF_TARGET set to esp32'.format(os.path.abspath(expected_json_path)))

                if isinstance(target, list):
                    if idf_target in target:
                        target = idf_target
                    else:
                        raise ValueError('IDF_TARGET set to {}, not in decorator target value'.format(idf_target))
                else:
                    if idf_target != target:
                        raise ValueError('IDF_TARGET set to {}, not equal to decorator target value'.format(idf_target))

            dut_dict = kwargs['dut_dict']
            if target not in dut_dict:
                raise Exception('target can only be {%s}' % ', '.join(dut_dict.keys()))

            dut = dut_dict[target]
            try:
                # try to config the default behavior of erase nvs
                dut.ERASE_NVS = kwargs['erase_nvs']
            except AttributeError:
                pass

            env_config['dut'] = dut
            env_inst = Env.Env(**env_config)

            # prepare for xunit test results
            junit_file_path = env_inst.app_cls.get_log_folder(env_config["test_suite_name"])
            junit_test_case = JunitReport.create_test_case(case_info["ID"])
            result = False

            try:
                Utility.console_log("starting running test: " + test_func.__name__, color="green")
                # execute test function
                test_func(env_inst, extra_data)
                # if finish without exception, test result is True
                result = True
            except Exception as e:
                # handle all the exceptions here
                traceback.print_exc()
                # log failure
                junit_test_case.add_failure_info(str(e) + ":\r\n" + traceback.format_exc())
            finally:
                # do close all DUTs, if result is False then print DUT debug info
                close_errors = env_inst.close(dut_debug=(not result))
                # We have a hook in DUT close, allow DUT to raise error to fail test case.
                # For example, we don't allow DUT exception (reset) during test execution.
                # We don't want to implement in exception detection in test function logic,
                # as we need to add it to every test case.
                # We can implement it in DUT receive thread,
                # and raise exception in DUT close to fail test case if reset detected.
                if close_errors:
                    for error in close_errors:
                        junit_test_case.add_failure_info(str(error))
                    result = False
                if not case_info["junit_report_by_case"]:
                    JunitReport.test_case_finish(junit_test_case)

            # end case and output result
            JunitReport.output_report(junit_file_path)

            if result:
                Utility.console_log("Test Succeed: " + test_func.__name__, color="green")
            else:
                Utility.console_log(("Test Fail: " + test_func.__name__), color="red")
            return result

        handle_test.case_info = case_info
        handle_test.test_method = True
        return handle_test

    return test