poky/meta/lib/oeqa/core/case.py

#
# Copyright (C) 2016 Intel Corporation
#
# SPDX-License-Identifier: MIT
#

import base64
import zlib
import unittest

from oeqa.core.exception import OEQAMissingVariable

def _validate_td_vars(td, td_vars, type_msg):
    if td_vars:
        for v in td_vars:
            if not v in td:
                raise OEQAMissingVariable("Test %s need %s variable but"\
                        " isn't into td" % (type_msg, v))

class OETestCase(unittest.TestCase):
    # TestContext and Logger instance set by OETestLoader.
    tc = None
    logger = None

    # td has all the variables needed by the test cases
    # is the same across all the test cases.
    td = None

    # td_vars has the variables needed by a test class
    # or test case instance, if some var isn't into td a
    # OEQAMissingVariable exception is raised
    td_vars = None

    @classmethod
    def _oeSetUpClass(clss):
        _validate_td_vars(clss.td, clss.td_vars, "class")
        if hasattr(clss, 'setUpHooker') and callable(getattr(clss, 'setUpHooker')):
            clss.setUpHooker()
        clss.setUpClassMethod()

    @classmethod
    def _oeTearDownClass(clss):
        clss.tearDownClassMethod()

    def _oeSetUp(self):
        try:
            for d in self.decorators:
                d.setUpDecorator()
        except:
            for d in self.decorators:
                d.tearDownDecorator()
            raise
        self.setUpMethod()

    def _oeTearDown(self):
        for d in self.decorators:
            d.tearDownDecorator()
        self.tearDownMethod()

class OEPTestResultTestCase:
    """
    Mix-in class to provide functions to make interacting with extraresults for
    the purposes of storing ptestresult data.
    """
    @staticmethod
    def _compress_log(log):
        logdata = log.encode("utf-8") if isinstance(log, str) else log
        logdata = zlib.compress(logdata)
        logdata = base64.b64encode(logdata).decode("utf-8")
        return {"compressed" : logdata}

    def ptest_rawlog(self, log):
        if not hasattr(self, "extraresults"):
            self.extraresults = {"ptestresult.sections" : {}}
        self.extraresults["ptestresult.rawlogs"] = {"log" : self._compress_log(log)}

    def ptest_section(self, section, duration = None, log = None, logfile = None, exitcode = None):
        if not hasattr(self, "extraresults"):
            self.extraresults = {"ptestresult.sections" : {}}

        sections = self.extraresults.get("ptestresult.sections")
        if section not in sections:
            sections[section] = {}

        if log is not None:
            sections[section]["log"] = self._compress_log(log)
        elif logfile is not None:
            with open(logfile, "rb") as f:
                sections[section]["log"] = self._compress_log(f.read())

        if duration is not None:
            sections[section]["duration"] = duration
        if exitcode is not None:
            sections[section]["exitcode"] = exitcode

    def ptest_result(self, section, test, result):
        if not hasattr(self, "extraresults"):
            self.extraresults = {"ptestresult.sections" : {}}

        sections = self.extraresults.get("ptestresult.sections")
        if section not in sections:
            sections[section] = {}
        resultname = "ptestresult.{}.{}".format(section, test)
        self.extraresults[resultname] = {"status" : result}