# resulttool - common library/utility functions
#
# Copyright (c) 2019, Intel Corporation.
# Copyright (c) 2019, Linux Foundation
#
# SPDX-License-Identifier: GPL-2.0-only
#
import os
import base64
import zlib
import json
import scriptpath
import copy
import urllib.request
import posixpath
scriptpath.add_oe_lib_path()
flatten_map = {
"oeselftest": [],
"runtime": [],
"sdk": [],
"sdkext": [],
"manual": []
}
regression_map = {
"oeselftest": ['TEST_TYPE', 'MACHINE'],
"runtime": ['TESTSERIES', 'TEST_TYPE', 'IMAGE_BASENAME', 'MACHINE', 'IMAGE_PKGTYPE', 'DISTRO'],
"sdk": ['TESTSERIES', 'TEST_TYPE', 'IMAGE_BASENAME', 'MACHINE', 'SDKMACHINE'],
"sdkext": ['TESTSERIES', 'TEST_TYPE', 'IMAGE_BASENAME', 'MACHINE', 'SDKMACHINE'],
"manual": ['TEST_TYPE', 'TEST_MODULE', 'IMAGE_BASENAME', 'MACHINE']
}
store_map = {
"oeselftest": ['TEST_TYPE'],
"runtime": ['TEST_TYPE', 'DISTRO', 'MACHINE', 'IMAGE_BASENAME'],
"sdk": ['TEST_TYPE', 'MACHINE', 'SDKMACHINE', 'IMAGE_BASENAME'],
"sdkext": ['TEST_TYPE', 'MACHINE', 'SDKMACHINE', 'IMAGE_BASENAME'],
"manual": ['TEST_TYPE', 'TEST_MODULE', 'MACHINE', 'IMAGE_BASENAME']
}
def is_url(p):
"""
Helper for determining if the given path is a URL
"""
return p.startswith('http://') or p.startswith('https://')
extra_configvars = {'TESTSERIES': ''}
#
# Load the json file and append the results data into the provided results dict
#
def append_resultsdata(results, f, configmap=store_map, configvars=extra_configvars):
if type(f) is str:
if is_url(f):
with urllib.request.urlopen(f) as response:
data = json.loads(response.read().decode('utf-8'))
url = urllib.parse.urlparse(f)
testseries = posixpath.basename(posixpath.dirname(url.path))
else:
with open(f, "r") as filedata:
data = json.load(filedata)
testseries = os.path.basename(os.path.dirname(f))
else:
data = f
for res in data:
if "configuration" not in data[res] or "result" not in data[res]:
raise ValueError("Test results data without configuration or result section?")
for config in configvars:
if config == "TESTSERIES" and "TESTSERIES" not in data[res]["configuration"]:
data[res]["configuration"]["TESTSERIES"] = testseries
continue
if config not in data[res]["configuration"]:
data[res]["configuration"][config] = configvars[config]
testtype = data[res]["configuration"].get("TEST_TYPE")
if testtype not in configmap:
raise ValueError("Unknown test type %s" % testtype)
testpath = "/".join(data[res]["configuration"].get(i) for i in configmap[testtype])
if testpath not in results:
results[testpath] = {}
results[testpath][res] = data[res]
#
# Walk a directory and find/load results data
# or load directly from a file
#
def load_resultsdata(source, configmap=store_map, configvars=extra_configvars):
results = {}
if is_url(source) or os.path.isfile(source):
append_resultsdata(results, source, configmap, configvars)
return results
for root, dirs, files in os.walk(source):
for name in files:
f = os.path.join(root, name)
if name == "testresults.json":
append_resultsdata(results, f, configmap, configvars)
return results
def filter_resultsdata(results, resultid):
newresults = {}
for r in results:
for i in results[r]:
if i == resultsid:
newresults[r] = {}
newresults[r][i] = results[r][i]
return newresults
def strip_ptestresults(results):
newresults = copy.deepcopy(results)
#for a in newresults2:
# newresults = newresults2[a]
for res in newresults:
if 'result' not in newresults[res]:
continue
if 'ptestresult.rawlogs' in newresults[res]['result']:
del newresults[res]['result']['ptestresult.rawlogs']
if 'ptestresult.sections' in newresults[res]['result']:
for i in newresults[res]['result']['ptestresult.sections']:
if 'log' in newresults[res]['result']['ptestresult.sections'][i]:
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"))
data = zlib.decompress(data)
return data.decode("utf-8", errors='ignore')
return None
def generic_get_log(sectionname, results, section):
if sectionname not in results:
return None
if section not in results[sectionname]:
return None
ptest = results[sectionname][section]
if 'log' not in ptest:
return None
return decode_log(ptest['log'])
def ptestresult_get_log(results, section):
return generic_get_log('ptestresuls.sections', results, section)
def generic_get_rawlogs(sectname, results):
if sectname not in results:
return None
if 'log' not in results[sectname]:
return None
return decode_log(results[sectname]['log'])
def ptestresult_get_rawlogs(results):
return generic_get_rawlogs('ptestresult.rawlogs', results)
def save_resultsdata(results, destdir, fn="testresults.json", ptestjson=False, ptestlogs=False):
for res in results:
if res:
dst = destdir + "/" + res + "/" + fn
else:
dst = destdir + "/" + fn
os.makedirs(os.path.dirname(dst), exist_ok=True)
resultsout = results[res]
if not ptestjson:
resultsout = strip_ptestresults(results[res])
with open(dst, 'w') as f:
f.write(json.dumps(resultsout, sort_keys=True, indent=4))
for res2 in results[res]:
if ptestlogs and 'result' in results[res][res2]:
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(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(sectionlog)
def git_get_result(repo, tags, configmap=store_map):
git_objs = []
for tag in tags:
files = repo.run_cmd(['ls-tree', "--name-only", "-r", tag]).splitlines()
git_objs.extend([tag + ':' + f for f in files if f.endswith("testresults.json")])
def parse_json_stream(data):
"""Parse multiple concatenated JSON objects"""
objs = []
json_d = ""
for line in data.splitlines():
if line == '}{':
json_d += '}'
objs.append(json.loads(json_d))
json_d = '{'
else:
json_d += line
objs.append(json.loads(json_d))
return objs
# Optimize by reading all data with one git command
results = {}
for obj in parse_json_stream(repo.run_cmd(['show'] + git_objs + ['--'])):
append_resultsdata(results, obj, configmap=configmap)
return results
def test_run_results(results):
"""
Convenient generator function that iterates over all test runs that have a
result section.
Generates a tuple of:
(result json file path, test run name, test run (dict), test run "results" (dict))
for each test run that has a "result" section
"""
for path in results:
for run_name, test_run in results[path].items():
if not 'result' in test_run:
continue
yield path, run_name, test_run, test_run['result']