1 files changed, 173 insertions, 0 deletions
diff --git a/util/zephyr_to_resultdb.py b/util/zephyr_to_resultdb.py
new file mode 100755
index 0000000000..48bfe151e4
--- /dev/null
+++ b/util/zephyr_to_resultdb.py
@@ -0,0 +1,173 @@
+#!/usr/bin/env python3
+# Copyright 2022 The ChromiumOS Authors
+# Use of this source code is governed by a BSD-style license that can be
+# found in the LICENSE file.
+""" Upload twister results to ResultDB
+
+    Usage:
+    $ rdb stream -new -realm chromium:public -- ./util/zephyr_to_resultdb.py
+      --results=twister-out/twister.json --upload=True
+"""
+
+import argparse
+import base64
+import json
+import os
+
+import requests  # pylint: disable=import-error
+
+
+def translate_status(status):
+    """Translates ZTEST status to ResultDB status"""
+    ret_status = "SKIP"
+
+    if status == "passed":
+        ret_status = "PASS"
+    elif status == "failed":
+        ret_status = "FAIL"
+    elif status in ["skipped", "filtered"]:
+        ret_status = "SKIP"
+
+    return ret_status
+
+
+def translate_expected(status):
+    """Translates ZTEST status to ResultDB expected"""
+    flag = False
+
+    if status in ["passed", "filtered"]:
+        flag = True
+
+    return flag
+
+
+def translate_duration(testcase):
+    """Translates ZTEST execution_time to ResultDB duration"""
+    time = testcase.get("execution_time")
+    if not time:
+        return None
+
+    return f"{time}ms"
+
+
+def testcase_summary(testcase):
+    """Translates ZTEST testcase to ResultDB summaryHtml"""
+    html = "<p>None</p>"
+
+    if (
+        "log" in testcase
+        or "reason" in testcase
+        or translate_status(testcase["status"]) == "SKIP"
+    ):
+        html = (
+            '<p><text-artifact artifact-id="artifact-content-in-request"></p>'
+        )
+
+    return html
+
+
+def testcase_artifact(testcase):
+    """Translates ZTEST testcase to ResultDB artifact"""
+    artifact = "Unknown"
+
+    if "log" in testcase and testcase["log"]:
+        artifact = testcase["log"]
+    elif "reason" in testcase and testcase["reason"]:
+        artifact = testcase["reason"]
+    elif testcase["status"] == "filtered":
+        artifact = "filtered"
+    elif testcase["status"] == "skipped":
+        artifact = "skipped"
+
+    return base64.b64encode(artifact.encode())
+
+
+def testcase_to_result(testsuite, testcase):
+    """Translates ZTEST testcase to ResultDB format"""
+    result = {
+        "testId": testcase["identifier"],
+        "status": translate_status(testcase["status"]),
+        "expected": translate_expected(testcase["status"]),
+        "summaryHtml": testcase_summary(testcase),
+        "artifacts": {
+            "artifact-content-in-request": {
+                "contents": testcase_artifact(testcase),
+            }
+        },
+        # TODO(b/239952573) Add all test configs as tags
+        "tags": [
+            {"key": "category", "value": "ChromeOS/EC"},
+            {"key": "platform", "value": testsuite["platform"]},
+        ],
+        "duration": translate_duration(testcase),
+        "testMetadata": {"name": testcase["identifier"]},
+    }
+
+    return result
+
+
+def json_to_resultdb(result_file):
+    """Translates Twister json test report to ResultDB format"""
+    with open(result_file) as file:
+        data = json.load(file)
+        results = []
+
+        for testsuite in data["testsuites"]:
+            for testcase in testsuite["testcases"]:
+                if testcase["status"]:
+                    results.append(testcase_to_result(testsuite, testcase))
+
+        file.close()
+
+    return results
+
+
+class BytesEncoder(json.JSONEncoder):
+    """Encoder for ResultDB format"""
+
+    def default(self, obj):
+        if isinstance(obj, bytes):
+            return obj.decode("utf-8")
+        return json.JSONEncoder.default(self, obj)
+
+
+def upload_results(results):
+    """Upload results to ResultDB"""
+    with open(os.environ["LUCI_CONTEXT"]) as file:
+        sink = json.load(file)["result_sink"]
+
+    # Uploads all test results at once.
+    res = requests.post(
+        url="http://%s/prpc/luci.resultsink.v1.Sink/ReportTestResults"
+        % sink["address"],
+        headers={
+            "Content-Type": "application/json",
+            "Accept": "application/json",
+            "Authorization": "ResultSink %s" % sink["auth_token"],
+        },
+        data=json.dumps({"testResults": results}, cls=BytesEncoder),
+    )
+    res.raise_for_status()
+
+
+def main():
+    """main"""
+    # Set up argument parser.
+    parser = argparse.ArgumentParser(
+        description=("Upload Zephyr Twister test results to ResultDB")
+    )
+    parser.add_argument("--results")
+    parser.add_argument("--upload", default=False)
+    args = parser.parse_args()
+
+    if args.results:
+        print("Converting:", args.results)
+        rdb_results = json_to_resultdb(args.results)
+        if args.upload:
+            upload_results(rdb_results)
+    else:
+        raise Exception("Missing test result file for conversion")
+
+
+if __name__ == "__main__":
+    main()