<commit_before>#!/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 datetime
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, base_tags):
    """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"]},
    }

    for (key, value) in base_tags:
        result["tags"].append({"key": key, "value": value})

    return result


def create_base_tags(data):
    """Creates base tags needed for Testhaus"""
    base_tags = []

    queued_time = datetime.datetime.fromisoformat(
        data["environment"]["run_date"]
    )
    base_tags.append(
        ("queued_time", queued_time.strftime("%Y-%m-%d %H:%M:%S.%f UTC"))
    )

    base_tags.append(("zephyr_version", data["environment"]["zephyr_version"]))
    base_tags.append(("board", data["environment"]["os"]))
    base_tags.append(("toolchain", data["environment"]["toolchain"]))

    return base_tags


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 = []
        base_tags = create_base_tags(data)

        for testsuite in data["testsuites"]:
            for testcase in testsuite["testcases"]:
                if testcase["status"]:
                    results.append(
                        testcase_to_result(testsuite, testcase, base_tags)
                    )

        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"<commit_msg>twister: Add config to tags for tests<commit_after>#!/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 datetime
import json
import os
import pathlib
import re

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, base_tags, config_tags):
    """Translates ZTEST testcase to ResultDB format
    See TestResult type in
    https://crsrc.org/i/go/src/go.chromium.org/luci/resultdb/sink/proto/v1/test_result.proto
    """
    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),
            }
        },
        "tags": [
            {"key": "platform", "value": testsuite["platform"]},
        ],
        "duration": translate_duration(testcase),
        "testMetadata": {"name": testcase["identifier"]},
    }

    for (key, value) in base_tags:
        result["tags"].append({"key": key, "value": value})

    for (key, value) in config_tags:
        result["tags"].append({"key": key.lower(), "value": value})

    return result


def get_testsuite_config_tags(twister_dir, testsuite):
    """Creates config tags from the testsuite"""
    config_tags = []
    suite_path = f"{twister_dir}/{testsuite['platform']}/{testsuite['name']}"
    autoconf_h = f"{suite_path}/zephyr/include/generated/autoconf.h"

    if pathlib.Path(autoconf_h).exists():
        with open(autoconf_h) as file:
            lines = file.readlines()

            for line in lines:
                result = re.search(r"#define\s*(\w+)\s*(.+$)", line)
                config_tags.append((result.group(1), result.group(2)))
    else:
        print(f"Can't find config file for {testsuite['name']}")

    return config_tags


def create_base_tags(data):
    """Creates base tags needed for Testhaus"""
    base_tags = []

    queued_time = datetime.datetime.fromisoformat(
        data["environment"]["run_date"]
    )
    base_tags.append(
        ("queued_time", queued_time.strftime("%Y-%m-%d %H:%M:%S.%f UTC"))
    )

    base_tags.append(("zephyr_version", data["environment"]["zephyr_version"]))
    base_tags.append(("board", data["environment"]["os"]))
    base_tags.append(("toolchain", data["environment"]["toolchain"]))

    return base_tags


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 = []
        base_tags = create_base_tags(data)

        for testsuite in data["testsuites"]:
            config_tags = get_testsuite_config_tags(
                os.path.dirname(result_file), testsuite
            )
            for testcase in testsuite["testcases"]:
                if testcase["status"]:
                    results.append(
                        testcase_to_result(
                            testsuite, testcase, base_tags, config_tags
                        )
                    )

        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"
{'fraction_non_alphanumeric': 0.1317928130813669, 'fraction_numerical': 0.00372074806619015, 'mean_word_length': 2.7998511904761907, 'pattern_counts': {'":': 42, '<': 13, '<?xml version=': 0, '>': 13, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4410364', 'n_tokens_mistral': 3157, 'n_tokens_neox': 3027, 'n_words': 733}