xref: /openbmc/openbmc/poky/meta/lib/oeqa/core/case.py (revision c926e17c) 
1#
2# Copyright (C) 2016 Intel Corporation
3#
4# SPDX-License-Identifier: MIT
5#
6
7import base64
8import zlib
9import unittest
10
11from oeqa.core.exception import OEQAMissingVariable
12
13def _validate_td_vars(td, td_vars, type_msg):
14    if td_vars:
15        for v in td_vars:
16            if not v in td:
17                raise OEQAMissingVariable("Test %s need %s variable but"\
18                        " isn't into td" % (type_msg, v))
19
20class OETestCase(unittest.TestCase):
21    # TestContext and Logger instance set by OETestLoader.
22    tc = None
23    logger = None
24
25    # td has all the variables needed by the test cases
26    # is the same across all the test cases.
27    td = None
28
29    # td_vars has the variables needed by a test class
30    # or test case instance, if some var isn't into td a
31    # OEQAMissingVariable exception is raised
32    td_vars = None
33
34    @classmethod
35    def _oeSetUpClass(clss):
36        _validate_td_vars(clss.td, clss.td_vars, "class")
37        if hasattr(clss, 'setUpHooker') and callable(getattr(clss, 'setUpHooker')):
38            clss.setUpHooker()
39        clss.setUpClassMethod()
40
41    @classmethod
42    def _oeTearDownClass(clss):
43        clss.tearDownClassMethod()
44
45    def _oeSetUp(self):
46        try:
47            for d in self.decorators:
48                d.setUpDecorator()
49        except:
50            for d in self.decorators:
51                d.tearDownDecorator()
52            raise
53        self.setUpMethod()
54
55    def _oeTearDown(self):
56        for d in self.decorators:
57            d.tearDownDecorator()
58        self.tearDownMethod()
59
60class OEPTestResultTestCase:
61    """
62    Mix-in class to provide functions to make interacting with extraresults for
63    the purposes of storing ptestresult data.
64    """
65    @staticmethod
66    def _compress_log(log):
67        logdata = log.encode("utf-8") if isinstance(log, str) else log
68        logdata = zlib.compress(logdata)
69        logdata = base64.b64encode(logdata).decode("utf-8")
70        return {"compressed" : logdata}
71
72    def ptest_rawlog(self, log):
73        if not hasattr(self, "extraresults"):
74            self.extraresults = {"ptestresult.sections" : {}}
75        self.extraresults["ptestresult.rawlogs"] = {"log" : self._compress_log(log)}
76
77    def ptest_section(self, section, duration = None, log = None, logfile = None, exitcode = None):
78        if not hasattr(self, "extraresults"):
79            self.extraresults = {"ptestresult.sections" : {}}
80
81        sections = self.extraresults.get("ptestresult.sections")
82        if section not in sections:
83            sections[section] = {}
84
85        if log is not None:
86            sections[section]["log"] = self._compress_log(log)
87        elif logfile is not None:
88            with open(logfile, "rb") as f:
89                sections[section]["log"] = self._compress_log(f.read())
90
91        if duration is not None:
92            sections[section]["duration"] = duration
93        if exitcode is not None:
94            sections[section]["exitcode"] = exitcode
95
96    def ptest_result(self, section, test, result):
97        if not hasattr(self, "extraresults"):
98            self.extraresults = {"ptestresult.sections" : {}}
99
100        sections = self.extraresults.get("ptestresult.sections")
101        if section not in sections:
102            sections[section] = {}
103        resultname = "ptestresult.{}.{}".format(section, test)
104        self.extraresults[resultname] = {"status" : result}
105
106