# SPDX-License-Identifier: MIT
#
+import base64
+import zlib
import unittest
from oeqa.core.exception import OEQAMissingVariable
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")
+ 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, "r") 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}
+
import resulttool.resultutils as resultutils
def show_ptest(result, ptest, logger):
- if 'ptestresult.sections' in result:
- if ptest in result['ptestresult.sections'] and 'log' in result['ptestresult.sections'][ptest]:
- print(result['ptestresult.sections'][ptest]['log'])
- return 0
+ logdata = resultutils.ptestresult_get_log(result, ptest)
+ if logdata is not None:
+ print(logdata)
+ return 0
- print("ptest '%s' not found" % ptest)
+ print("ptest '%s' log not found" % ptest)
return 1
def show_reproducible(result, reproducible, logger):
print("reproducible '%s' not found" % reproducible)
return 1
-
def log(args, logger):
results = resultutils.load_resultsdata(args.source)
return 1
for _, run_name, _, r in resultutils.test_run_results(results):
- if args.dump_ptest:
- if 'ptestresult.sections' in r:
- for name, ptest in r['ptestresult.sections'].items():
- if 'log' in ptest:
- dest_dir = args.dump_ptest
- if args.prepend_run:
- dest_dir = os.path.join(dest_dir, run_name)
-
- os.makedirs(dest_dir, exist_ok=True)
-
- dest = os.path.join(dest_dir, '%s.log' % name)
- print(dest)
- with open(dest, 'w') as f:
- f.write(ptest['log'])
+ if args.dump_ptest and 'ptestresult.sections' in r:
+ for name, ptest in r['ptestresult.sections'].items():
+ logdata = resultutils.ptestresult_get_log(r, name)
+ if logdata is not None:
+ dest_dir = args.dump_ptest
+ if args.prepend_run:
+ dest_dir = os.path.join(dest_dir, run_name)
+
+ os.makedirs(dest_dir, exist_ok=True)
+ dest = os.path.join(dest_dir, '%s.log' % name)
+ print(dest)
+ with open(dest, 'w') as f:
+ f.write(logdata)
if args.raw_ptest:
- if 'ptestresult.rawlogs' in r:
- print(r['ptestresult.rawlogs']['log'])
+ rawlog = resultutils.ptestresult_get_rawlogs(r)
+ if rawlog is not None:
+ print(rawlog)
else:
print('Raw ptest logs not found')
return 1
#
import os
+import base64
+import zlib
import json
import scriptpath
import copy
del newresults[res]['result']['ptestresult.sections'][i]['log']
return newresults
+def decode_log(logdata):
+ if isinstance(logdata, str):
+ return logdata
+ elif isinstance(logdata, dict):
+ if "compressed" in logdata:
+ data = logdata.get("compressed")
+ data = base64.b64decode(data.encode("utf-8"))
+ return zlib.decompress(data).decode("utf-8")
+ return None
+
+def ptestresult_get_log(results, section):
+ if 'ptestresult.sections' not in results:
+ return None
+ if section not in results['ptestresult.sections']:
+ return None
+
+ ptest = results['ptestresult.sections'][section]
+ if 'log' not in ptest:
+ return None
+ return decode_log(ptest['log'])
+
+def ptestresult_get_rawlogs(results):
+ if 'ptestresult.rawlogs' not in results:
+ return None
+ if 'log' not in results['ptestresult.rawlogs']:
+ return None
+ return decode_log(results['ptestresult.rawlogs']['log'])
+
def save_resultsdata(results, destdir, fn="testresults.json", ptestjson=False, ptestlogs=False):
for res in results:
if res:
f.write(json.dumps(resultsout, sort_keys=True, indent=4))
for res2 in results[res]:
if ptestlogs and 'result' in results[res][res2]:
- if 'ptestresult.rawlogs' in results[res][res2]['result']:
+ seriesresults = results[res][res2]['result']
+ rawlogs = ptestresult_get_rawlogs(seriesresults)
+ if rawlogs is not None:
with open(dst.replace(fn, "ptest-raw.log"), "w+") as f:
- f.write(results[res][res2]['result']['ptestresult.rawlogs']['log'])
- if 'ptestresult.sections' in results[res][res2]['result']:
- for i in results[res][res2]['result']['ptestresult.sections']:
- if 'log' in results[res][res2]['result']['ptestresult.sections'][i]:
+ f.write(rawlogs)
+ if 'ptestresult.sections' in seriesresults:
+ for i in seriesresults['ptestresult.sections']:
+ sectionlog = ptestresult_get_log(seriesresults, i)
+ if sectionlog is not None:
with open(dst.replace(fn, "ptest-%s.log" % i), "w+") as f:
- f.write(results[res][res2]['result']['ptestresult.sections'][i]['log'])
+ f.write(sectionlog)
def git_get_result(repo, tags):
git_objs = []
