automated/utils/post-to-squad.py

#!/usr/bin/env python

import argparse
import datetime
import decimal
import json
import logging
import os
import requests
from collections import OrderedDict


def parse_args():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "-r",
        "--result-file",
        dest="result_file",
        required=True,
        default="./result.json",
        help="Specify test result file.",
    )
    parser.add_argument(
        "-a",
        "--attachment",
        dest="attachment",
        action="append",
        help="Specify attachment file.",
    )
    parser.add_argument(
        "-t",
        "--team",
        dest="team",
        required=True,
        help='Team identifier. Defaults to "erp"',
    )
    parser.add_argument(
        "-p",
        "--project",
        dest="project",
        help="Project identifier. Defaults to the name of the Linux distribution.",
    )
    parser.add_argument(
        "-b", "--build", dest="build", required=True, help="Build identifier."
    )
    parser.add_argument(
        "-e",
        "--test-env",
        dest="test_env",
        help="Environment identifier. Defaults to board name.",
    )
    parser.add_argument(
        "-u",
        "--url",
        dest="url",
        default="https://qa-reports.linaro.org",
        help="Dashboard URL. Defaults to https://qa-reports.linaro.org.",
    )
    parser.add_argument(
        "-v",
        "--verbose",
        action="store_true",
        dest="verbose",
        default=True,
        help="Set log level.",
    )

    args = parser.parse_args()
    return args


def squad_result(results):
    squad_tests = OrderedDict()
    squad_metrics = OrderedDict()
    for result in results:
        for metric in result["metrics"]:
            key = "%s/%s" % (result["name"], metric["test_case_id"])
            if not metric["measurement"]:
                # Collect pass/fail test results.
                squad_tests[key] = metric["result"]
            else:
                # Collect performance test results.
                try:
                    measurement = decimal.Decimal(metric["measurement"])
                    squad_metrics[key] = float(measurement)
                except decimal.InvalidOperation:
                    logger.info("Invalid measurement: %s" % metric["measurement"])
                    logger.info("Skipped adding: %s" % metric)
    assert squad_tests or squad_metrics, "No valid result found!"
    return (squad_tests, squad_metrics)


def squad_metadata(results):
    test_plan = list(set(i["test_plan"] for i in results))
    test_version = list(set(i["version"] for i in results))

    assert len(test_plan) == 1, "More then one test plan found!"
    assert len(test_version) == 1, "More then one test version found!"

    squad_metadata = OrderedDict()
    test_plan = test_plan[0]
    test_plan_name = os.path.splitext(os.path.basename(test_plan))[0]
    squad_metadata["job_id"] = "{}_{}".format(
        test_plan_name, datetime.datetime.utcnow().isoformat()
    )
    squad_metadata["test_plan"] = test_plan
    squad_metadata["test_version"] = test_version[0]
    for key, value in results[-1]["environment"].items():
        if key != "packages":
            squad_metadata[key] = value
    return squad_metadata


def main():
    auth_token = os.environ.get("SQUAD_AUTH_TOKEN")
    assert auth_token, "SQUAD_AUTH_TOKEN not provided in environment"

    with open(args.result_file, "r") as f:
        results = json.load(f)
    metadata = squad_metadata(results)
    tests, metrics = squad_result(results)

    files = [
        ("metadata", json.dumps(metadata)),
        ("tests", json.dumps(tests)),
        ("metrics", json.dumps(metrics)),
        ("attachment", open(args.result_file, "rb")),
    ]
    if args.attachment is not None:
        for item in args.attachment:
            if os.path.exists(item):
                logger.info("Adding {} to attachment list...".format(item))
                files.append(tuple(["attachment", open(item, "rb")]))
            else:
                logger.info("Attachment %s Not found" % args.attachment)
                logger.info("Skipped uploading %s" % args.attachment)
    logger.debug("Data to post: %s" % files)

    project = args.project or metadata["linux_distribution"]
    test_env = args.test_env or metadata["board_name"]
    url = "{}/api/submit/{}/{}/{}/{}".format(
        args.url, args.team, project, args.build, test_env
    )
    logger.info("Posting to {}".format(url))

    headers = {"Auth-Token": auth_token}
    r = requests.post(url, headers=headers, files=files)
    print(r.text)


if __name__ == "__main__":
    args = parse_args()

    logger = logging.getLogger("post-to-squad")
    logger.setLevel(logging.INFO)
    if args.verbose:
        logger.setLevel(logging.DEBUG)
    ch = logging.StreamHandler()
    ch.setLevel(logging.DEBUG)
    formatter = logging.Formatter(
        "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
    )
    ch.setFormatter(formatter)
    logger.addHandler(ch)

    main()