1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
|
#
# 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):
for d in self.decorators:
d.setUpDecorator()
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}
|
