#
|
# 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}
|
